--- /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