--- /dev/null
+package org.opendaylight.controller.sal.connector.remoterpc.router.zeromq;
+
+import java.io.IOException;
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.opendaylight.controller.sal.connector.api.RpcRouter;
+import org.opendaylight.controller.sal.connector.remoterpc.router.zeromq.Message.MessageType;
+import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
+import org.opendaylight.controller.sal.core.api.Broker.RpcRegistration;
+import org.opendaylight.controller.sal.core.api.RpcImplementation;
+import org.opendaylight.controller.sal.core.api.RpcRegistrationListener;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.common.RpcError;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
+import org.zeromq.ZMQ;
+
+/**
+ * ZeroMq based implementation of RpcRouter
+ * TODO:
+ * 1. Make it multi VM aware
+ * 2. Make rpc request handling async and non-blocking. Note zmq socket is not thread safe
+ * 3. sendRpc() should use connection pooling
+ * 4. Read properties from config file using existing(?) ODL properties framework
+ */
+public class ZeroMqRpcRouter implements RpcRouter<QName, QName, InstanceIdentifier, Object> {
+
+ private ExecutorService serverPool;
+ private static ExecutorService handlersPool;
+
+ private Map<RpcRouter.RouteIdentifier<QName, QName, InstanceIdentifier>, String> routingTable;
+
+ private ProviderSession brokerSession;
+
+ private ZMQ.Context context;
+ private ZMQ.Socket publisher;
+ private ZMQ.Socket subscriber;
+ private ZMQ.Socket replySocket;
+
+ private static ZeroMqRpcRouter _instance = new ZeroMqRpcRouter();
+
+ private final RpcFacade facade = new RpcFacade();
+ private final RpcListener listener = new RpcListener();
+
+ private final String localIp = getLocalIpAddress();
+
+ private String pubPort = System.getProperty("pub.port");// port on which announcements are sent
+ private String subPort = System.getProperty("sub.port");// other controller's pub port
+ private String pubIp = System.getProperty("pub.ip"); // other controller's ip
+ private String rpcPort = System.getProperty("rpc.port");// port on which RPC messages are received
+
+ //Prevent instantiation
+ private ZeroMqRpcRouter() {
+ }
+
+ public static ZeroMqRpcRouter getInstance() {
+ return _instance;
+ }
+
+ public void start() {
+ context = ZMQ.context(2);
+ publisher = context.socket(ZMQ.PUB);
+ int ret = publisher.bind("tcp://*:" + pubPort);
+ System.out.println(Thread.currentThread().getName() + " Return(publish bind) :[" + ret + "]");
+ // serverPool = Executors.newSingleThreadExecutor();
+ serverPool = Executors.newCachedThreadPool();
+ handlersPool = Executors.newCachedThreadPool();
+ routingTable = new ConcurrentHashMap<RpcRouter.RouteIdentifier<QName, QName, InstanceIdentifier>, String>();
+
+ // Start listening for announce and rpc messages
+ serverPool.execute(receive());
+
+ brokerSession.addRpcRegistrationListener(listener);
+
+ Set<QName> currentlySupported = brokerSession.getSupportedRpcs();
+ for (QName rpc : currentlySupported) {
+ listener.onRpcImplementationAdded(rpc);
+ }
+
+ }
+
+ public void stop() {
+ if (handlersPool != null)
+ handlersPool.shutdown();
+ if (serverPool != null)
+ serverPool.shutdown();
+ if (publisher != null) {
+ publisher.setLinger(0);
+ publisher.close();
+ }
+ if (replySocket != null) {
+ replySocket.setLinger(0);
+ replySocket.close();
+ }
+ if (subscriber != null) {
+ subscriber.setLinger(0);
+ subscriber.close();
+ }
+ if (context != null)
+ context.term();
+
+ }
+
+ private Runnable receive() {
+ return new Runnable() {
+ public void run() {
+ try {
+ // Bind to RPC reply socket
+ replySocket = context.socket(ZMQ.REP);
+ replySocket.bind("tcp://*:" + rpcPort);
+
+ // Bind to publishing controller
+ subscriber = context.socket(ZMQ.SUB);
+ subscriber.connect("tcp://" + pubIp + ":" + subPort);
+ System.out.println(Thread.currentThread().getName() + "-Subscribing at[" + "tcp://"
+ + pubIp + ":" + subPort + "]");
+
+ //subscribe for announcements
+ //TODO: Message type would be changed. Update this
+ subscriber.subscribe(Message.serialize(Message.MessageType.ANNOUNCE));
+
+ // Poller enables listening on multiple sockets using a single thread
+ ZMQ.Poller poller = new ZMQ.Poller(2);
+ poller.register(replySocket, ZMQ.Poller.POLLIN);
+ poller.register(subscriber, ZMQ.Poller.POLLIN);
+ System.out.println(Thread.currentThread().getName() + "-Start Polling");
+
+ //TODO: Add code to restart the thread after exception
+ while (!Thread.currentThread().isInterrupted()) {
+
+ poller.poll();
+
+ if (poller.pollin(0)) {
+ handleRpcCall();
+ }
+ if (poller.pollin(1)) {
+ handleAnnouncement();
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ replySocket.setLinger(0);
+ replySocket.close();
+ subscriber.setLinger(0);
+ subscriber.close();
+ }
+ };
+ }
+
+ /**
+ * @throws IOException
+ * @throws ClassNotFoundException
+ */
+ private void handleAnnouncement() throws IOException, ClassNotFoundException {
+ System.out.println("\n" + Thread.currentThread().getName() + "-Received message");
+ Message.MessageType topic = (MessageType) Message.deserialize(subscriber.recv());
+ System.out.println("Topic:[" + topic + "]");
+
+ if (subscriber.hasReceiveMore()) {
+ try {
+ Message m = (Message) Message.deserialize(subscriber.recv());
+ System.out.println(m);
+ // TODO: check on msg type or topic. Both
+ // should be same. Need to normalize.
+ if (Message.MessageType.ANNOUNCE == m.getType())
+ updateRoutingTable(m);
+ } catch (IOException | ClassNotFoundException e) {
+ e.printStackTrace();
+ }
+ }
+
+ }
+
+ /**
+ * @throws InterruptedException
+ * @throws ExecutionException
+ */
+ private void handleRpcCall() throws InterruptedException, ExecutionException {
+ try {
+ Message req = parseMessage(replySocket);
+
+ System.out.println("Received RPC request [" + req + "]");
+
+ // Call broker to process the message then reply
+ Future<RpcResult<CompositeNode>> rpc = brokerSession.rpc(
+ (QName) req.getRoute().getType(), (CompositeNode) req.getPayload());
+
+ RpcResult<CompositeNode> result = rpc.get();
+
+ Message response = new Message.MessageBuilder()
+ .type(MessageType.RESPONSE)
+ .sender(localIp + ":" + rpcPort)
+ .route(req.getRoute())
+ //.payload(result) TODO: enable and test
+ .build();
+
+ replySocket.send(Message.serialize(response));
+
+ System.out.println("Sent RPC response [" + response + "]");
+
+ } catch (IOException ex) {
+ //TODO: handle exception and send error codes to caller
+ System.out.println("Rpc request could not be handled" + ex);
+ }
+ }
+
+
+ @Override
+ public Future<RpcReply<Object>> sendRpc(
+ final RpcRequest<QName, QName, InstanceIdentifier, Object> input) {
+
+ return handlersPool.submit(new Callable<RpcReply<Object>>() {
+
+ @Override
+ public RpcReply<Object> call() {
+ ZMQ.Socket requestSocket = context.socket(ZMQ.REQ);
+
+ // TODO pick the ip and port from routing table based on routing identifier
+ requestSocket.connect("tcp://" + pubIp + ":5554");
+
+ Message requestMessage = new Message.MessageBuilder()
+ .type(MessageType.REQUEST)
+ .sender(localIp + ":" + rpcPort)
+ .route(input.getRoutingInformation())
+ .payload(input.getPayload())
+ .build();
+
+ RpcReply<Object> reply = null;
+
+ try {
+ System.out.println("\n\nRPC Request [" + requestMessage + "]");
+
+ requestSocket.send(Message.serialize(requestMessage));
+ final Message resp = parseMessage(requestSocket);
+
+ System.out.println("\n\nRPC Response [" + resp + "]");
+
+ reply = new RpcReply<Object>() {
+
+ @Override
+ public Object getPayload() {
+ return resp.getPayload();
+ }
+ };
+ } catch (IOException ex) {
+ // TODO: Pass exception back to the caller
+ System.out.println("Error in RPC send. Input could not be serialized[" + input + "]");
+ }
+
+ return reply;
+ }
+ });
+ }
+
+ /**
+ * TODO: Remove this implementation and use RoutingTable implementation to send announcements
+ * Publishes a notice to other controllers in the cluster
+ *
+ * @param notice
+ */
+ public void publish(final Message notice) {
+ Runnable task = new Runnable() {
+ public void run() {
+
+ System.out.println(
+ Thread.currentThread().getName() + "-Publisher started at port[" + pubPort + "]");
+
+ try {
+
+ System.out.println(
+ Thread.currentThread().getName() + "-Sending announcement[" + notice + "]");
+
+ publisher.sendMore(Message.serialize(Message.MessageType.ANNOUNCE));
+ publisher.send(Message.serialize(notice));
+
+ } catch (IOException ex) {
+ System.out.println("Error in publishing");
+ ex.printStackTrace();
+ }
+ System.out.println(Thread.currentThread().getName() + "-Published message[" + notice
+ + "]");
+
+ }
+ };
+ handlersPool.execute(task);
+ }
+
+ /**
+ * Finds IPv4 address of the local VM
+ * TODO: This method is non-deterministic. There may be more than one IPv4 address. Cant say which
+ * address will be returned. Read IP from a property file or enhance the code to make it deterministic.
+ * Should we use IP or hostname?
+ *
+ * @return
+ */
+ private String getLocalIpAddress() {
+ String hostAddress = null;
+ Enumeration e = null;
+ try {
+ e = NetworkInterface.getNetworkInterfaces();
+ } catch (SocketException e1) {
+ e1.printStackTrace();
+ }
+ while (e.hasMoreElements()) {
+
+ NetworkInterface n = (NetworkInterface) e.nextElement();
+
+ Enumeration ee = n.getInetAddresses();
+ while (ee.hasMoreElements()) {
+ InetAddress i = (InetAddress) ee.nextElement();
+ if ((i instanceof Inet4Address) && (i.isSiteLocalAddress()))
+ hostAddress = i.getHostAddress();
+ }
+ }
+ return hostAddress;
+
+ }
+
+ /**
+ * TODO: Change to use external routing table implementation
+ *
+ * @param msg
+ */
+ private void updateRoutingTable(Message msg) {
+ routingTable.put(msg.getRoute(), msg.getSender());
+ RpcRouter.RouteIdentifier<QName, QName, InstanceIdentifier> route = msg.getRoute();
+
+ // Currently only registers rpc implementation.
+ // TODO: do registration for instance based routing
+ QName rpcType = route.getType();
+ RpcRegistration registration = brokerSession.addRpcImplementation(rpcType, facade);
+ }
+
+ /**
+ * @param socket
+ * @return
+ */
+ private Message parseMessage(ZMQ.Socket socket) {
+
+ Message msg = null;
+ try {
+ byte[] bytes = socket.recv();
+ System.out.println("Received bytes:[" + bytes.length + "]");
+ msg = (Message) Message.deserialize(bytes);
+ } catch (Throwable t) {
+ System.out.println("Caught exception");
+ t.printStackTrace();
+ }
+ return msg;
+ }
+
+ private class RpcFacade implements RpcImplementation {
+
+ @Override
+ public Set<QName> getSupportedRpcs() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
+
+ RouteIdentifierImpl routeId = new RouteIdentifierImpl();
+ routeId.setType(rpc);
+
+ RpcRequestImpl request = new RpcRequestImpl();
+ request.setRouteIdentifier(routeId);
+ request.setPayload(input);
+
+ final Future<RpcReply<Object>> ret = sendRpc(request);
+
+ //TODO: Review result handling
+ RpcResult<CompositeNode> result = new RpcResult<CompositeNode>() {
+ @Override
+ public boolean isSuccessful() {
+ try {
+ ret.get();
+ } catch (InterruptedException | ExecutionException e) {
+ e.printStackTrace();
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public CompositeNode getResult() {
+ return null;
+ }
+
+ @Override
+ public Collection<RpcError> getErrors() {
+ return Collections.EMPTY_LIST;
+ }
+ };
+ return result;
+ }
+ }
+
+ /**
+ * Listener for rpc registrations
+ */
+ private class RpcListener implements RpcRegistrationListener {
+
+ @Override
+ public void onRpcImplementationAdded(QName name) {
+ System.out.println("In ZeroMQ Rpc Listener onRpcImplementationAdded()");
+
+ RouteIdentifierImpl routeId = new RouteIdentifierImpl();
+ routeId.setType(name);
+
+ //TODO: Make notice immutable and change message type
+ Message notice = new Message.MessageBuilder()
+ .type(MessageType.ANNOUNCE)
+ .sender("tcp://" + localIp + ":" + rpcPort)
+ .route(routeId)
+ .build();
+
+ publish(notice);
+ }
+
+ @Override
+ public void onRpcImplementationRemoved(QName name) {
+ // TODO: send a rpc-deregistrtation notice
+
+ }
+ }
+
+ public void setBrokerSession(ProviderSession session) {
+ this.brokerSession = session;
+
+ }
+
+}
Code Review / controller.git / blobdiff