+++ /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 java.net.Inet4Address;
-import java.net.InetAddress;
-import java.net.NetworkInterface;
-import java.net.SocketException;
-import java.util.Enumeration;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.FutureTask;
-import java.util.concurrent.TimeUnit;
-
-import org.opendaylight.controller.sal.core.api.Broker.ProviderSession;
-import org.opendaylight.yangtools.yang.common.QName;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.zeromq.ZMQ;
-
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-
-/**
- * ZeroMq based implementation of RpcRouter.
- */
-public class ServerImpl implements RemoteRpcServer {
-
- private final Logger _logger = LoggerFactory.getLogger(ServerImpl.class);
-
- private ExecutorService serverPool;
- protected ServerRequestHandler handler;
-
- private Set<QName> remoteServices;
- private ProviderSession brokerSession;
- private ZMQ.Context context;
-
- private final String HANDLER_INPROC_ADDRESS = "inproc://rpc-request-handler";
- private final int HANDLER_WORKER_COUNT = 2;
- private final int HWM = 200;//high water mark on sockets
- private volatile State status = State.STOPPED;
-
- private String serverAddress;
- private final int port;
-
- public static enum State {
- STARTING, STARTED, STOPPED;
- }
-
- public ServerImpl(int port) {
- this.port = port;
- }
-
- public State getStatus() {
- return this.status;
- }
-
- public Optional<ServerRequestHandler> getHandler() {
- return Optional.fromNullable(this.handler);
- }
-
- public void setBrokerSession(ProviderSession session) {
- this.brokerSession = session;
- }
-
- public Optional<ProviderSession> getBrokerSession() {
- return Optional.fromNullable(this.brokerSession);
- }
-
- public Optional<ZMQ.Context> getZmqContext() {
- return Optional.fromNullable(this.context);
- }
-
- public String getServerAddress() {
- return serverAddress;
- }
-
- public String getHandlerAddress() {
- return HANDLER_INPROC_ADDRESS;
- }
-
- /**
- *
- */
- public void start() {
- Preconditions.checkState(State.STOPPED == this.getStatus(),
- "Remote RPC Server is already running");
-
- status = State.STARTING;
- _logger.debug("Remote RPC Server is starting...");
-
- String hostIpAddress = findIpAddress();
-
- //Log and silently die as per discussion in the bug (bug-362)
- //https://bugs.opendaylight.org/show_bug.cgi?id=362
- //
- // A tracking enhancement defect (bug-366) is created to properly fix this issue
- //https://bugs.opendaylight.org/show_bug.cgi?id=366
- //checkState(hostIpAddress != null, "Remote RPC Server could not acquire host ip address");
-
- if (hostIpAddress == null) {
- _logger.error("Remote RPC Server could not acquire host ip address. Stopping...");
- stop();
- return;
- }
-
- this.serverAddress = new StringBuilder(hostIpAddress).
- append(":").
- append(port).
- toString();
-
- context = ZMQ.context(1);
- remoteServices = new HashSet<QName>();//
- serverPool = Executors.newSingleThreadExecutor();//main server thread
- serverPool.execute(receive()); // Start listening rpc requests
-
- status = State.STARTED;
- _logger.info("Remote RPC Server started [{}]", getServerAddress());
- }
-
- public void stop(){
- close();
- }
-
- /**
- *
- */
- @Override
- public void close() {
-
- if (State.STOPPED == this.getStatus()) return; //do nothing
-
- if (serverPool != null)
- serverPool.shutdown();
-
- closeZmqContext();
-
- status = State.STOPPED;
- _logger.info("Remote RPC Server stopped");
- }
-
- /**
- * Closes ZMQ Context. It tries to gracefully terminate the context. If
- * termination takes more than 5 seconds, its forcefully shutdown.
- */
- private void closeZmqContext() {
- ExecutorService exec = Executors.newSingleThreadExecutor();
- FutureTask<?> zmqTermination = new FutureTask<Void>(new Runnable() {
-
- @Override
- public void run() {
- try {
- if (context != null)
- context.term();
- _logger.debug("ZMQ Context terminated gracefully!");
- } catch (Exception e) {
- _logger.debug("ZMQ Context termination threw exception [{}]. Continuing shutdown...", e);
- }
- }
- }, null);
-
- exec.execute(zmqTermination);
-
- try {
- zmqTermination.get(5L, TimeUnit.SECONDS);
- } catch (Exception e) {/*ignore and continue with shutdown*/}
-
- exec.shutdownNow();
- }
-
- /**
- * Main listener thread that spawns {@link ServerRequestHandler} as workers.
- *
- * @return
- */
- private Runnable receive() {
- return new Runnable() {
-
- @Override
- public void run() {
- Thread.currentThread().setName("remote-rpc-server");
- _logger.debug("Remote RPC Server main thread starting...");
-
- //socket clients connect to (frontend)
- ZMQ.Socket clients = context.socket(ZMQ.ROUTER);
-
- //socket RequestHandlers connect to (backend)
- ZMQ.Socket workers = context.socket(ZMQ.DEALER);
-
- try (SocketPair capturePair = new SocketPair();
- ServerRequestHandler requestHandler = new ServerRequestHandler(context,
- brokerSession,
- HANDLER_WORKER_COUNT,
- HANDLER_INPROC_ADDRESS,
- getServerAddress());) {
-
- handler = requestHandler;
- clients.setHWM(HWM);
- clients.bind("tcp://*:" + port);
- workers.setHWM(HWM);
- workers.bind(HANDLER_INPROC_ADDRESS);
- //start worker threads
- _logger.debug("Remote RPC Server worker threads starting...");
- requestHandler.start();
- //start capture thread
- // handlerPool.execute(new CaptureHandler(capturePair.getReceiver()));
- // Connect work threads to client threads via a queue
- ZMQ.proxy(clients, workers, null);//capturePair.getSender());
-
- } catch (Exception e) {
- _logger.debug("Unhandled exception [{}, {}]", e.getClass(), e.getMessage());
- } finally {
- if (clients != null) clients.close();
- if (workers != null) workers.close();
- _logger.info("Remote RPC Server stopped");
- }
- }
- };
- }
-
- /**
- * 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 findIpAddress() {
- Enumeration<?> e = null;
- try {
- e = NetworkInterface.getNetworkInterfaces();
- } catch (SocketException e1) {
- _logger.error("Failed to get list of interfaces", e1);
- return null;
- }
- while (e.hasMoreElements()) {
-
- NetworkInterface n = (NetworkInterface) e.nextElement();
-
- Enumeration<?> ee = n.getInetAddresses();
- while (ee.hasMoreElements()) {
- InetAddress i = (InetAddress) ee.nextElement();
- _logger.debug("Trying address {}", i);
- if ((i instanceof Inet4Address) && (!i.isLoopbackAddress())) {
- String hostAddress = i.getHostAddress();
- _logger.debug("Settled on host address {}", hostAddress);
- return hostAddress;
- }
- }
- }
-
- _logger.error("Failed to find a suitable host address");
- return null;
- }
-
-}
Code Review / controller.git / blobdiff