import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public void run() {
    LOGGER.info("Started ZMQ pusher");

    pull.connect("tcp://127.0.0.1:"+ port);

    while (!Thread.interrupted() || stopped.get()) {
        String m = null;
        try {
            m = pull.recvStr();
            InputHandler.handleMessage(m);
        } catch (ZMQException ex) {
            LOGGER.error("ZMQ error in KV7/8 processing", ex);
        } catch (Exception e) {
            LOGGER.error("Error in KV7/8 processing", e);
            metrics.increaseBucketValue("kv78turbo.messages.errors", ChronoUnit.HOURS);
            if (m != null) {
                LOGGER.debug("Got message {}", m);
            }
        }
    }

    LOGGER.debug("Processing task is interrupted");
    disconnect();
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
/**
 * Receive and parse multipart message.
 *
 * @param socket    zmq socket to receive message on
 * @param flag      parameters passed to ZMsg.recvMsg()
 * @return reply    null if no message is received
 *
 * @throws ProtocolException    RequestDispatcher protocol violated
 * @throws ZmqEtermException    when context is closed while receiving messages
 * @throws IOException          other IOErrors (encompasses Protocol and ZmqEtermException)
 */
public static TransferWrapper recvMessage(ZMQ.Socket socket, int flag) throws ProtocolException {
    {
        try {
            ZMsg message = ZMsg.recvMsg(socket, flag);

            if (message == null || message.size() <= 2 ) {
                // message.size() check works around bug in jeromq:
                // https://github.com/zeromq/jeromq/commit/750b5f408ab2e925d17de700eda3e16185b770fc
                return null;
            }

            return new TransferWrapper(message);

        } catch (ZMQException e) {
            if (e.getErrorCode() == ZMQ.Error.ETERM.getCode()){
                log.debug("Received ETERM.");
                throw new ZmqEtermException(e);
            } else {
                throw e;
            }
        }
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
/**
 * Print messages to output stream.
 * Socket gets closed on ETERM.
 */
@Override
public void run() {
    try {
        log.info("Starting print loop.");
        String msg = null;
        while (true) {
            msg = socket.recvStr();
            log.debug("Received " + msg);
            outputStream.println(msg);
        }
    } catch (ZMQException e) {
        log.info("Catched ETERM. Closing socket. ");
        socket.close();
        outputStream.flush();
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
/**
 * Push every message received on to the ring buffer
 * until signaled to stop
 *
 * Critical Path! Any additional work here will impact
 * total throughput!
 */
@Override
public void run() {
    logger.debug("starting server socket");
    while(socketListening) {
        // receive bytes representing a metric to be saved
        byte[] receivedBytes = null;
        try {
            receivedBytes = serverSocket.recv(0);
        } catch (ZMQException e) {
            logger.error(e.getMessage());
            close();
        }

        // publish metric to ring buffer
        long seq = ringBuffer.next();
        ByteArrayHolder metricHolder = ringBuffer.get(seq);
        metricHolder.setValue(receivedBytes);
        ringBuffer.publish(seq);
    }

    logger.debug("zeromq listening socket exiting");
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
protected ZMQ.Socket createConnectableSocketWithStandardSettings() {
    ZMQ.Context zmqContext = context.getZMQContext();
    ZMQ.Socket socket;
    try {
        socket = zmqContext.socket(this.getSocketType().getType());
        socket.setLinger(getLinger());
        socket.setSndHWM(getSendHighWaterMark());
        socket.setRcvHWM(getReceiveHighWaterMark());
        socket.setReceiveTimeOut(getSocketSpec().receiveTimeout);
        socket.setSendTimeOut(getSocketSpec().sendTimeout);
        if (this.getIdentity() != null && this.getIdentity().length > 0) {
            socket.setIdentity(this.getIdentity());
        }
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
    return socket;
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
protected ZMQ.Socket createBindableSocketWithStandardSettings() {
    ZMQ.Context zmqContext = context.getZMQContext();
    ZMQ.Socket socket;
    try {
        socket = zmqContext.socket(this.getSocketType().getType());
        socket.setLinger(this.getLinger());
        socket.setRcvHWM(this.getReceiveHighWaterMark());
        socket.setSndHWM(this.getSendHighWaterMark());
        socket.setReceiveTimeOut(getSocketSpec().receiveTimeout);
        socket.setSendTimeOut(getSocketSpec().sendTimeout);
        if (this.getIdentity() != null && this.getIdentity().length > 0) {
            socket.setIdentity(this.getIdentity());
        }
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
    return socket;
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
synchronized private RPCResponse[] doSend(String json) throws IOException {
    LOGGER.info("JSON Req: " + json);
    zmqSocket.send(json);
    byte[] msg = zmqSocket.recv(0);
    if (msg == null) {
        int errNumber = zmqSocket.base().errno();
        throw new ZMQException("Unable to receive message from socket", errNumber);
    }
    String response = new String(msg);
    LOGGER.info("JSON Resp: " + response);
    
    return new ObjectMapper().readValue(response, RPCResponse[].class);
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
private String call(String json) {
    LOGGER.info("JSON Req: " + json);
    byte[] msg = transport.sendJson(json);
    if (msg == null) {
        int errNumber = transport.getSocket().base().errno();
        String errMsg = "Unable to receive message from socket";
        ZMQException zmqException = new ZMQException(errMsg, errNumber);
        LOGGER.error(errMsg, zmqException);
        throw zmqException;
    }
    String response = new String(msg);
    LOGGER.info("JSON Resp: " + response);
    return response;
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
public static void sendMessage(ZMQ.Socket socket, TransferWrapper transferWrapper) {
    try {
        boolean rc = transferWrapper.toMessage().send(socket);
        if (!rc) throw new ZMQException.IOException(new IOException("Error sending message"));
    } catch (ZMQException e) {
        if (e.getErrorCode() == ZMQ.Error.ETERM.getCode()){
            log.debug("Received ETERM.");
            throw new ZmqEtermException(e);
        } else {
            throw e;
        }
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
public Router(final String addr) {

    new Thread() {

      @Override
      public void run() {
        Context ctx = ZMQ.context(1);
        Socket router = ctx.socket(ZMQ.ROUTER);
        LOGGER.info("Binding reply socket to " + addr);
        router.bind(addr);

        while (!Thread.currentThread().isInterrupted()) {
          try {
            String id = router.recvStr();
            router.sendMore(id);
            router.recv();
            String request = router.recvStr();
            router.sendMore("");

            LOGGER.info("Handling request " + request);
            String response = handle(request);

            LOGGER.info("Sending response");
            router.send(response);
          } catch (Exception e) {
            LOGGER.warn("Error handling request", e);

            if (e instanceof ZMQException) {
              LOGGER.error("Will close reply socket");
              break;
            }
          }
        }
        router.close();
        ctx.close();
        LOGGER.info("Closed reply socket " + addr);
      }

    }.start();
  }
 

import org.zeromq.ZMQException; //导入依赖的package包/类
public static byte[] receiveMessage(ZMQ.Socket zmqSocket) {
    try {
        return zmqSocket.recv(0);
    } catch (ZMQException e) {
        if(e.getErrorCode() == ZMQURIConstants.ERR_ETERM) {
            LOG.log(Level.FINE, "ZeroMQ context terminated. Closing socket...");
            zmqSocket.close();
            return null;
        }
        else {
            throw e;
        }
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
/**
 * Wrap an underlying ZMQException in the appropriate higher level exception.
 * 
 * @param thrown The underlying ZMQException
 * @return A new exception, which wraps a ZMQException
 */
public static ZMQRuntimeException wrap(ZMQException thrown) {
    if (isContextTerminated(thrown)) {
        return new ContextTerminatedException(thrown);
    } else if (isInvalidSocket(thrown)) {
        return new InvalidSocketException(thrown);
    } else {
        return new ZMQRuntimeException(thrown);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public void poll(long timeoutMillis) {
    try {
        int numberOfObjects = poller.poll(timeoutMillis);
        if (numberOfObjects == 0) {
            return;
        }

        // simulate ETERM to make JeroMQ act like jzmq, which no longer returns -1
        if (numberOfObjects < 0) {
            throw new ZMQException("Simulated ETERM error", (int) ZMQ.Error.ETERM.getCode());
        }
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }

    for (int index = 0; index < pollables.size(); index++) {
        Pollable pollable = pollables.get(index);
        PollListener listener = listeners.get(index);
        if (pollable != null) {
            if (poller.pollin(index))
                listener.handleIn(pollable);
            if (poller.pollout(index))
                listener.handleOut(pollable);
            if (poller.pollerr(index))
                listener.handleError(pollable);
        }
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public void run() {
    try {
        backgroundable.run(context, pipe);
    } catch (ZMQException ex) {
        if (!ZMQExceptions.isContextTerminated(ex)) {
            throw ZMQExceptions.wrap(ex);
        }
    }
    log.debug("Background thread {} has shut down", Thread.currentThread().getName());
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public byte[] receive() {
    checkClosed();
    try {
        return socket.recv(0);
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public int receiveByteBuffer(ByteBuffer buf, MessageFlag flag) {
    checkClosed();
    try {
        return socket.recvByteBuffer(buf, flag.getFlag());
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public boolean hasMoreToReceive() {
    checkClosed();
    try {
        return socket.hasReceiveMore();
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public boolean send(byte[] buf, int offset, int length, MessageFlag flag) {
    checkClosed();
    try {
        return socket.send(buf, offset, length, flag.getFlag());
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
@Override
public boolean sendByteBuffer(ByteBuffer buf, MessageFlag flag) {
    checkClosed();
    try {
        return socket.sendByteBuffer(buf, flag.getFlag()) >= 0;
    } catch (ZMQException ex) {
        throw ZMQExceptions.wrap(ex);
    }
}
 

import org.zeromq.ZMQException; //导入依赖的package包/类
boolean createSockets() {
    ctx = new ZContext();
    String ip = (String) connectionData.get("ip");
    String transport = (String) connectionData.get("transport");
    try {
        // http://jupyter-client.readthedocs.org/en/latest/messaging.html#heartbeat-for-kernels 
        //
        //     Clients send ping messages on a REQ socket, which are echoed right back from the Kernel’s 
        //     REP socket. These are simple bytestrings, not full JSON messages described above.
        Heartbeat = ctx.createSocket(ZMQ.REP);
        Heartbeat.bind(String.format("%s://%s:%s",
                transport, ip, connectionData.get("hb_port")
        ));

        // http://jupyter-client.readthedocs.org/en/latest/messaging.html#introduction
        // Shell: this single ROUTER socket allows multiple incoming connections from frontends, and 
        //        this is the socket where requests for code execution, object information, prompts, etc. 
        //        are made to the kernel by any frontend. 
        String shellAddress = String.format("%s://%s:%s",
                transport, ip, connectionData.get("shell_port"));
        Shell = ctx.createSocket(ZMQ.ROUTER);
        Shell.bind(shellAddress);

        // Control: This channel is identical to Shell, but operates on a separate socket, to allow 
        //          important messages to avoid queueing behind execution requests (e.g. shutdown or abort).
        Control = ctx.createSocket(ZMQ.ROUTER);
        Control.bind(String.format("%s://%s:%s",
                transport, ip, connectionData.get("control_port")
        ));

        // IOPub: this socket is the ‘broadcast channel’ where the kernel publishes all side effects 
        //        (stdout, stderr, etc.) as well as the requests coming from any client over the shell socket 
        //        send its own requests on the stdin socket. 
        IOPub = ctx.createSocket(ZMQ.PUB);
        IOPub.bind(String.format("%s://%s:%s",
                transport, ip, connectionData.get("iopub_port")
        ));

        Stdin = ctx.createSocket(ZMQ.ROUTER);
        Stdin.connect(String.format("%s://%s:%s",
                transport, ip, connectionData.get("stdin_port")
        ));
    } catch (ZMQException e) {
        closeSockets();
        Logger.getLogger(Session.class.getName()).log(Level.SEVERE, null, e);
    }

    sockets = new ZMQ.Poller(4);
    sockets.register(Control, ZMQ.Poller.POLLIN);
    sockets.register(Heartbeat, ZMQ.Poller.POLLIN);
    sockets.register(Shell, ZMQ.Poller.POLLIN);
    sockets.register(Stdin, ZMQ.Poller.POLLIN);
    return true;
}