--- /dev/null
+package org.opendaylight.controller.sal.connector.zeromq;
+
+import org.opendaylight.controller.sal.connector.api.RpcRouter;
+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.RpcResult;
+import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier;
+import org.zeromq.ZMQ;
+
+import java.io.IOException;
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.util.*;
+import java.util.concurrent.*;
+
+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 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
+
+
+ private ZeroMqRpcRouter() {
+ }
+
+ public static ZeroMqRpcRouter getInstance() {
+ return _instance;
+ }
+
+ public void start() {
+ context = ZMQ.context(2);
+ serverPool = Executors.newSingleThreadExecutor();
+ 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("Subscribing at[" + "tcp://" + pubIp + ":" + subPort + "]");
+
+ subscriber.subscribe(Message.serialize(Message.MessageType.ANNOUNCE));
+
+ // Initialize poll set
+ ZMQ.Poller poller = new ZMQ.Poller(2);
+ poller.register(replySocket, ZMQ.Poller.POLLIN);
+ poller.register(subscriber, ZMQ.Poller.POLLIN);
+
+ while (!Thread.currentThread().isInterrupted()) {
+
+ poller.poll(250);
+ //TODO: Fix this
+ if (poller.pollin(0)) {
+ //receive rpc request and reply
+ try {
+ Message req = parseMessage(replySocket);
+ Message resp = new Message();
+ //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();
+ resp.setType(Message.MessageType.RESPONSE);
+ resp.setSender(getLocalIpAddress() + ":" + rpcPort);
+ resp.setRoute(req.getRoute());
+ resp.setPayload(result.isSuccessful());
+ replySocket.send(Message.serialize(resp));
+
+ } catch (IOException ex) {// | ClassNotFoundException ex) {
+ System.out.println("Rpc request could not be handled" + ex);
+ }
+ }
+ if (poller.pollin(1)) {
+ //get subscription and update routing table
+ //try {
+ Message.MessageType topic = (Message.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(); //To change body of catch statement use File | Settings | File Templates.
+ }
+ }
+//
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ replySocket.setLinger(0);
+ replySocket.close();
+ subscriber.setLinger(0);
+ subscriber.close();
+ }
+ };
+ }
+
+ private void updateRoutingTable(Message msg) {
+ routingTable.put(msg.getRoute(), msg.getSender());
+ RpcRouter.RouteIdentifier<QName, QName, InstanceIdentifier> route = msg.getRoute();
+ QName rpcType = route.getType();
+ System.out.println("Routing Table\n" + routingTable);
+
+ RpcRegistration registration = brokerSession.addRpcImplementation(rpcType, facade);
+ }
+
+ private Message parseMessage(ZMQ.Socket socket) {
+ //Message m = new Message();
+ //socket.setReceiveBufferSize(40000);
+ 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;
+ /*m.setType((Message.MessageType) Message.deserialize(socket.recv()));
+
+ if (socket.hasReceiveMore()) {
+ m.setSender((String) Message.deserialize(socket.recv()));
+ }
+ if (socket.hasReceiveMore()) {
+ m.setRoute((RouteIdentifier) Message.deserialize(socket.recv()));
+ }
+ if (socket.hasReceiveMore()) {
+ m.setPayload(Message.deserialize(socket.recv()));
+ }
+ return m;*/
+ }
+
+ @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);
+ Message req = new Message();
+ Message resp = null;
+ RpcReplyImpl reply = new RpcReplyImpl();
+ requestSocket.connect((String) routingTable.get(input.getRoutingInformation().getRoute()));
+
+ req.setType(Message.MessageType.REQUEST);
+ req.setSender(getLocalIpAddress() + ":" + rpcPort);
+ req.setRoute(input.getRoutingInformation());
+ req.setPayload(input.getPayload());
+ try {
+ requestSocket.send(Message.serialize(req));
+ resp = parseMessage(requestSocket);
+ reply.setPayload(resp.getPayload());
+ } catch (IOException ex) {//| ClassNotFoundException ex) {
+ //Log and ignore
+ System.out.println("Error in RPC send. Input could not be serialized[" + input + "]");
+ }
+
+ return reply;
+ }
+ });
+ }
+
+ public void publish(final Message message) {
+ Runnable task = new Runnable() {
+ public void run() {
+ // Bind to publishing port
+ publisher = context.socket(ZMQ.PUB);
+ publisher.bind("tcp://*:" + pubPort);
+ System.out.println("Publisher started at port[" + pubPort + "]");
+ try {
+ Message outMessage = new Message();
+ outMessage.setType(Message.MessageType.ANNOUNCE);
+ outMessage.setSender("tcp://" + getLocalIpAddress() + ":" + rpcPort);
+ outMessage.setRoute(message.getRoute());
+
+ System.out.println("Sending announcement[" + outMessage + "]");
+ publisher.sendMore(Message.serialize(Message.MessageType.ANNOUNCE));
+ publisher.send(Message.serialize(outMessage));
+
+ } catch (IOException ex) {
+ //Log and ignore
+ System.out.println("Error in publishing");
+ ex.printStackTrace();
+ }
+ System.out.println("Published message[" + message + "]");
+ publisher.close();
+ }
+ };
+ handlersPool.execute(task);
+ }
+
+ 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;
+
+
+ }
+
+
+ private class RpcFacade implements RpcImplementation {
+
+
+ @Override
+ public Set<QName> getSupportedRpcs() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public RpcResult<CompositeNode> invokeRpc(QName rpc, CompositeNode input) {
+
+ RpcRequestImpl request = new RpcRequestImpl();
+ RouteIdentifierImpl routeId = new RouteIdentifierImpl();
+ routeId.setContext(null);
+ routeId.setRoute(null);
+ routeId.setType(rpc);
+
+ request.setRouteIdentifier(routeId);
+ request.setPayload(input);
+ // Create message
+
+ Future<org.opendaylight.controller.sal.connector.api.RpcRouter.RpcReply<Object>> ret = sendRpc(request);
+
+ return null;
+ }
+ }
+
+ private class RpcListener implements RpcRegistrationListener {
+
+ @Override
+ public void onRpcImplementationAdded(QName name) {
+
+ Message msg = new Message();
+ RouteIdentifierImpl routeId = new RouteIdentifierImpl();
+ routeId.setType(name);
+ msg.setRoute(routeId);
+ publish(msg);
+ }
+
+ @Override
+ public void onRpcImplementationRemoved(QName name) {
+ // TODO Auto-generated method stub
+
+ }
+ }
+
+ public void setBrokerSession(ProviderSession session) {
+ this.brokerSession = session;
+
+ }
+
+}
Code Review / controller.git / blobdiff