--- /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 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.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * A class encapsulating {@link ZMQ.Socket} of type {@link ZMQ.REQ}.
+ * It adds following capabilities:
+ * <li> Retry logic - Tries 3 times before giving up
+ * <li> Request times out after {@link TIMEOUT} property
+ * <li> The limitation of {@link ZMQ.REQ}/{@link ZMQ.REP} pair is that no 2 requests can be sent before
+ * the response for the 1st request is received. To overcome that, this socket queues all messages until
+ * the previous request has been responded.
+ */
+public class RpcSocket {
+
+ // Constants
+ public static final int TIMEOUT = 2000;
+ public static final int QUEUE_SIZE = 10;
+ public static final int NUM_RETRIES = 3;
+ private static final Logger log = LoggerFactory.getLogger(RpcSocket.class);
+
+ private ZMQ.Socket socket;
+ private ZMQ.Poller poller;
+ private String address;
+ private SocketState state;
+ private long sendTime;
+ private int retriesLeft;
+ private LinkedBlockingQueue<MessageWrapper> inQueue;
+
+
+ public RpcSocket(String address, ZMQ.Poller poller) {
+ this.socket = null;
+ this.state = new IdleSocketState();
+ this.sendTime = -1;
+ this.retriesLeft = NUM_RETRIES;
+ this.inQueue = new LinkedBlockingQueue<MessageWrapper>(QUEUE_SIZE);
+ this.address = address;
+ this.poller = poller;
+ createSocket();
+ }
+
+ public ZMQ.Socket getSocket() {
+ return socket;
+ }
+
+ public String getAddress() {
+ return address;
+ }
+
+ public int getRetriesLeft() {
+ return retriesLeft;
+ }
+
+ public void setRetriesLeft(int retriesLeft) {
+ this.retriesLeft = retriesLeft;
+ }
+
+ public SocketState getState() {
+ return state;
+ }
+
+ public void setState(SocketState state) {
+ this.state = state;
+ }
+
+ public int getQueueSize() {
+ return inQueue.size();
+ }
+
+ public MessageWrapper removeCurrentRequest() {
+ return inQueue.poll();
+ }
+
+ public boolean hasTimedOut() {
+ return (System.currentTimeMillis() - sendTime > RpcSocket.TIMEOUT);
+ }
+
+ public void send(MessageWrapper request) throws TimeoutException {
+ try {
+ boolean success = inQueue.offer(request, TIMEOUT, TimeUnit.MILLISECONDS);
+ if (!success) {
+ throw new TimeoutException("send :: Queue is full");
+ }
+ process();
+ }
+ catch (InterruptedException e) {
+ log.error("send : Thread interrupted while attempting to add request to inQueue", e);
+ }
+ }
+
+ public MessageWrapper receive() {
+ Message response = parseMessage();
+ MessageWrapper messageWrapper = inQueue.poll(); //remove the message from queue
+ MessageWrapper responseMessageWrapper = new MessageWrapper(response, messageWrapper.getReceiveSocket());
+
+ state = new IdleSocketState();
+ retriesLeft = NUM_RETRIES;
+ return responseMessageWrapper;
+ }
+
+ public void process() {
+ if (getQueueSize() > 0) //process if there's message in the queue
+ state.process(this);
+ }
+
+ // Called by IdleSocketState & BusySocketState
+ public void sendMessage() {
+ //Get the message from queue without removing it. For retries
+ MessageWrapper messageWrapper = inQueue.peek();
+ if (messageWrapper != null) {
+ Message message = messageWrapper.getMessage();
+ try {
+ socket.send(Message.serialize(message));
+ }
+ catch (IOException e) {
+ log.debug("Message send failed [{}]", message);
+ log.debug("Exception [{}]", e);
+ }
+ sendTime = System.currentTimeMillis();
+ }
+ }
+
+ public Message parseMessage() {
+ Message parsedMessage = null;
+ byte[] bytes = socket.recv();
+ log.debug("Received bytes:[{}]", bytes.length);
+ try {
+ parsedMessage = (Message)Message.deserialize(bytes);
+ }
+ catch (IOException|ClassNotFoundException e) {
+ log.debug("parseMessage : Deserializing received bytes failed", e);
+ }
+
+ return parsedMessage;
+ }
+
+ public void recycleSocket() {
+ close();
+ }
+
+ public void close() {
+ socket.setLinger(10);
+ socket.close();
+ }
+
+ private void createSocket() {
+ socket = Context.getInstance().getZmqContext().socket(ZMQ.REQ);
+ socket.connect(address);
+ poller.register(socket, ZMQ.Poller.POLLIN);
+ state = new IdleSocketState();
+ }
+
+
+ /**
+ * Represents the state of a {@link org.opendaylight.controller.sal.connector.remoterpc.RpcSocket}
+ */
+ public static interface SocketState {
+
+ /* The processing actions to be performed in this state
+ */
+ public void process(RpcSocket socket);
+ }
+
+ /**
+ * Represents the idle state of a {@link org.opendaylight.controller.sal.connector.remoterpc.RpcSocket}
+ */
+ public static class IdleSocketState implements SocketState {
+
+ @Override
+ public void process(RpcSocket socket) {
+ socket.sendMessage();
+ socket.setState(new BusySocketState());
+ socket.setRetriesLeft(socket.getRetriesLeft()-1);
+ }
+ }
+
+ /**
+ * Represents the busy state of a {@link org.opendaylight.controller.sal.connector.remoterpc.RpcSocket}
+ */
+ public static class BusySocketState implements SocketState {
+
+ private static Logger log = LoggerFactory.getLogger(BusySocketState.class);
+
+ @Override
+ public void process(RpcSocket socket) {
+ if (socket.hasTimedOut()) {
+ if (socket.getRetriesLeft() > 0) {
+ log.debug("process : Request timed out, retrying now...");
+ socket.sendMessage();
+ socket.setRetriesLeft(socket.getRetriesLeft() - 1);
+ }
+ else {
+ // No more retries for current request, so stop processing the current request
+ MessageWrapper message = socket.removeCurrentRequest();
+ if (message != null) {
+ log.error("Unable to process rpc request [{}]", message);
+ socket.setState(new IdleSocketState());
+ socket.setRetriesLeft(NUM_RETRIES);
+ }
+ }
+ }
+ // Else no timeout, so allow processing to continue
+ }
+ }
+}
Code Review / controller.git / blobdiff