+++ /dev/null
-/*
- * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
- * This program and the accompanying materials are made available under the
- * terms of the Eclipse Public License v1.0 which accompanies this distribution,
- * and is available at http://www.eclipse.org/legal/epl-v10.html
- */
-
-package org.opendaylight.controller.sal.connector.remoterpc;
-
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.sal.connector.remoterpc.dto.Message;
-import org.opendaylight.controller.sal.connector.remoterpc.dto.MessageWrapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.zeromq.ZMQ;
-
-import java.io.IOException;
-import java.util.concurrent.TimeoutException;
-
-import static com.google.common.base.Preconditions.*;
-
-/**
- * Main server thread for sending requests.
- */
-public class Sender implements Runnable{
-
- private final static Logger _logger = LoggerFactory.getLogger(Sender.class);
- private final Client client;
-
-
-
-
- public Sender(Client client) {
- super();
- this.client = client;
- }
-
-@Override
- public void run() {
- _logger.info("Starting...");
-
- try (SocketManager socketManager = new SocketManager()){
- while (!Thread.currentThread().isInterrupted()) {
-
- //read incoming messages from blocking queue
- MessageWrapper request = pollForRequest();
-
- if (request != null) {
- processRequest(socketManager, request);
- }
-
- flushSockets(socketManager);
- pollForResponse(socketManager);
- processResponse(socketManager);
-
- }
- } catch(Exception t){
- _logger.error("Exception: [{}]", t);
- _logger.error("Stopping...");
- }
- }
-
- private void processResponse(SocketManager socketManager) {
- for (int i = 0; i < socketManager.getPoller().getSize(); i++) {
- // If any sockets get a response, process it
- if (socketManager.getPoller().pollin(i)) {
- Optional<RpcSocket> socket = socketManager.getManagedSocketFor(
- socketManager.getPoller().getItem(i).getSocket());
-
- checkState(socket.isPresent(), "Managed socket not found");
-
- MessageWrapper response = socket.get().receive();
- _logger.debug("Received rpc response [{}]", response.getMessage());
-
- //TODO: handle exception and introduce timeout on receiver side
- try {
- response.getReceiveSocket().send(Message.serialize(response.getMessage()));
- } catch (IOException e) {
- e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates.
- }
- }
- }
- }
-
- private void processRequest(SocketManager socketManager, MessageWrapper request) throws TimeoutException {
-
- if ((request.getMessage() == null) ||
- (request.getMessage().getRecipient() == null)) {
- //invalid message. log and drop
- _logger.error("Invalid request [{}]", request);
- return;
- }
-
- RpcSocket socket =
- socketManager.getManagedSocket(request.getMessage().getRecipient());
-
- socket.send(request);
- }
-
- private void flushSockets(SocketManager socketManager){
- for (RpcSocket socket : socketManager.getManagedSockets()){
- socket.process();
- }
- }
-
- private MessageWrapper pollForRequest(){
- return client.getRequestQueue().poll();
- }
-
- private void pollForResponse(SocketManager socketManager){
- try{
- socketManager.getPoller().poll(10); //poll every 10ms
- }catch (Throwable t) { /*Ignore and continue*/ }
- }
-}
-
-
-/*
-SCALA
-
-package org.opendaylight.controller.sal.connector.remoterpc
-
- import org.slf4j.{LoggerFactory, Logger}
- import scala.collection.JavaConverters._
- import scala.Some
- import org.opendaylight.controller.sal.connector.remoterpc.dto.{MessageWrapper, Message}
-*/
-/**
- * Main server thread for sending requests. This does not maintain any state. If the
- * thread dies, it will be restarted
- */
-/*class Sender extends Runnable {
- private val _logger: Logger = LoggerFactory.getLogger(Sender.this.getClass())
-
- override def run = {
- _logger.info("Sender starting...")
- val socketManager = new SocketManager()
-
- try {
- while (!Thread.currentThread().isInterrupted) {
- //read incoming messages from blocking queue
- val request: MessageWrapper = Client.requestQueue.poll()
-
- if (request != null) {
- if ((request.message != null) &&
- (request.message.getRecipient != null)) {
-
- val socket = socketManager.getManagedSocket(request.message.getRecipient)
- socket.send(request)
- } else {
- //invalid message. log and drop
- _logger.error("Invalid request [{}]", request)
- }
- }
-
- socketManager.getManagedSockets().asScala.map(s => s.process)
-
- // Poll all sockets for responses every 1 sec
- poll(socketManager)
-
- // If any sockets get a response, process it
- for (i <- 0 until socketManager.poller.getSize) {
- if (socketManager.poller.pollin(i)) {
- val socket = socketManager.getManagedSocketFor(socketManager.poller.getItem(i).getSocket)
-
- socket match {
- case None => //{
- _logger.error("Could not find a managed socket for zmq socket")
- throw new IllegalStateException("Could not find a managed socket for zmq socket")
- //}
- case Some(s) => {
- val response = s.receive()
- _logger.debug("Received rpc response [{}]", response.message)
- response.receiveSocket.send(Message.serialize(response.message))
- }
- }
- }
- }
-
- }
- } catch{
- case e:Exception => {
- _logger.debug("Sender stopping due to exception")
- e.printStackTrace()
- }
- } finally {
- socketManager.stop
- }
- }
-
- def poll(socketManager:SocketManager) = {
- try{
- socketManager.poller.poll(10)
- }catch{
- case t:Throwable => //ignore and continue
- }
- }
-}
-
-
-// def newThread(r: Runnable): Thread = {
-// val t = new RequestHandler()
-// t.setUncaughtExceptionHandler(new RequestProcessorExceptionHandler)
-// t
-// }
-
-
-
-/**
- * Restarts the request processing server in the event of unforeseen exceptions
- */
-//private class RequestProcessorExceptionHandler extends UncaughtExceptionHandler {
-// def uncaughtException(t: Thread, e: Throwable) = {
-// _logger.error("Exception caught during request processing [{}]", e)
-// _logger.info("Restarting request processor server...")
-// RequestProcessor.start()
-// }
Code Review / controller.git / blobdiff