public interface Backgroundable {
    /**
     * Run a background thread communicating over the given socket.
     * 
     * @param context The parent context
     * @param socket The socket used for communication
     */
    void run(Context context, Socket socket);

    /**
     * Called just prior to the socket being closed.
     */
    void onClose();
} public interface Context extends Closeable {
    /**
     * Create a ØMQ Socket of type SocketType
     * 
     * @param type socket type
     * @return A builder for constructing and connecting ØMQ Sockets
     */
    SocketBuilder buildSocket(SocketType type);

    /**
     * @return The ØMQ version, in a pretty-printed String.
     */
    String getVersionString();

    /**
     * @return The ØMQ version, in integer form.
     */
    int getFullVersion();

    /**
     * Create a new Poller, which will allow callback-based polling of Sockets.
     * 
     * @return A builder for constructing ØMQ Pollers
     * @see Pollable
     * @see PollerType
     */
    PollerBuilder buildPoller();

    /**
     * Create a new Reactor, which will allow event-driven and timer-based
     * polling of Sockets.
     * 
     * @return A builder for constructing a Reactor
     */
    ReactorBuilder buildReactor();

    /**
     * Create a new BinaryStarReactor, which will create one half of an HA-pair
     * with event-driven polling of a client Socket.
     * 
     * @return A builder for constructing a BinaryStarReactor
     */
    BinaryStarReactorBuilder buildBinaryStarReactor();

    /**
     * Create a ØMQ Socket, backed by a background agent that is connecting
     * to a BinaryStarReactor HA-pair.
     *
     * @return A builder for constructing connecting a BinaryStarReactor client Socket
     */
    BinaryStarSocketBuilder buildBinaryStarSocket();

    /**
     * Create a new CloneServer, which will create one half of an HA-pair
     * for distributing key/value data to clients.
     * 
     * @return A builder for constructing a CloneServer
     */
    CloneServerBuilder buildCloneServer();

    /**
     * Create a new CloneClient, connected to an HA-pair, for publishing key/value data
     * to anonymous peers.
     *
     * @return A builder for constructing a CloneClient
     */
    CloneClientBuilder buildCloneClient();

    /**
     * Create a new BeaconReactor, which will send and receive UDP beacons
     * on a broadcast address, with event-driven handling of received beacons.
     * 
     * @return A builder for constructing a BeaconReactor
     */
    BeaconReactorBuilder buildBeaconReactor();

    /**
     * Create a ØMQ Device of type DeviceType, which will bridge two networks
     * together using patterns for specific SocketTypes.
     *
     * @param deviceType The device type, specifying the pattern to use
     * @return A builder for constructing ØMQ Devices
     */
    DeviceBuilder buildDevice(DeviceType deviceType);

    /**
     * Create a new Pollable from the socket, with the requested options.
     * 
     * @param socket A socket to wrap for polling
     * @param options Polling options (IN, OUT, ERROR)
     * @return A new Pollable, for use with a Poller.
     */
    Pollable newPollable(Socket socket, PollerType... options);

    /**
     * Create a new Pollable from the socket, with the requested options.
     *
     * @param channel A channel to wrap for polling
     * @param options Polling options (IN, OUT, ERROR)
     * @return A new Pollable, for use with a Poller.
     */
    Pollable newPollable(SelectableChannel channel, PollerType... options);

    /**
     * Create a ØMQ proxy and start it up.  Returns when the context is closed.
     * 
     * @param frontEnd The front-end socket which will be proxied to/from the back-end
     * @param backEnd The back-end socket which will be proxied to/from the front-end
     */
    void proxy(Socket frontEnd, Socket backEnd);

    /**
     * Create a ØMQ proxy and start it up on another thread that exits when the
     * context is closed.
     *
     * @param frontEnd The front-end socket which will be proxied to/from the back-end
     * @param backEnd The back-end socket which will be proxied to/from the front-end
     */
    void forward(Socket frontEnd, Socket backEnd);

    /**
     * Alias of {@link #forward}.
     *
     * @param frontEnd The front-end socket which will be proxied to/from the back-end
     * @param backEnd The back-end socket which will be proxied to/from the front-end
     */
    void queue(Socket frontEnd, Socket backEnd);

    /**
     * Run a background thread with an inproc PAIR socket for communication.
     * 
     * @param backgroundable The task to be performed on the background thread
     * @return The inproc PAIR socket for communicating with the background thread
     */
    Socket fork(Backgroundable backgroundable);

    /**
     * Run a background thread using the given socket for communication.
     * 
     * @param socket The socket owned by the background thread
     * @param backgroundable The task to be performed on the background thread
     */
    void fork(Socket socket, Backgroundable backgroundable);

    /**
     * Create a new Context with the same underlying ØMQ context, with an empty
     * (separate) list of managed Sockets and Backgroundables.
     * <p>
     * The returned context will not attempt to terminate the underlying ØMQ
     * context when closed, and will only close managed sockets, etc.
     * 
     * @return A new context with the same underlying ØMQ context
     */
    Context shadow();

    /**
     * Close the context and any open sockets.
     */
    void close();

    /**
     * Asynchronously terminate the context without closing any open sockets,
     * forcing pollers and waiters to abort.
     */
    void terminate();
} public enum DeviceType {
    /**
     * ZMQ_STREAMER collects tasks from a set of pushers and forwards these to a set of pullers. You will generally use
     * this to bridge networks. Messages are fair-queued from pushers and load-balanced to pullers.
     */
    STREAMER(1),
    /**
     * ZMQ_FORWARDER collects messages from a set of publishers and forwards these to a set of subscribers. You will
     * generally use this to bridge networks, e.g. read on TCP unicast and forward on multicast.
     */
    FORWARDER(2),
    /**
     * ZMQ_QUEUE creates a shared queue that collects requests from a set of clients, and distributes these fairly among
     * a set of services. Requests are fair-queued from frontend connections and load-balanced between backend
     * connections. Replies automatically return to the client that made the original request.
     */
    QUEUE(3);

    private final int type;

    DeviceType(int type) {
        this.type = type;
    }

    /**
     * Device type connects a frontend socket to a backend socket.
     * 
     * @return device type
     */
    public int getType() {
        return type;
    }
} public interface Pollable {
    /**
     * Get the socket (if any) for this poll item.
     * 
     * @return The socket, or null if this is a SelectableChannel Pollable
     */
    Socket getSocket();

    /**
     * Get the SelectableChannel (if any) for this poll item.
     * 
     * @return The channel, or null if this is a Socket Pollable
     */
    SelectableChannel getChannel();

    /**
     * Get the options used for this poll item
     * 
     * @return The options used for this Pollable
     */
    EnumSet<PollerType> getOptions();
} public class ZMQExceptions {

    private ZMQExceptions() {}

    /**
     * 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);
        }
    }

    /**
     * Helper method to determine if error code is <code>ETERM</code>.
     * 
     * @param thrown A ZMQException, thrown by the library
     * @return true if the exception indicates ETERM, false otherwise
     */
    public static boolean isContextTerminated(ZMQException thrown) {
        return thrown.getErrorCode() == ZMQ.Error.ETERM.getCode();
    }

    /**
     * Helper method to determine if error code is <code>ENOTSOCK</code>.
     * 
     * @param thrown A ZMQException, thrown by the library
     * @return true if the exception indicates ENOTSOCK, false otherwise
     */
    public static boolean isInvalidSocket(ZMQException thrown) {
        return thrown.getErrorCode() == ZMQ.Error.ENOTSOCK.getCode()
            || thrown.getErrorCode() == ZMQ.Error.EAGAIN.getCode();
    }
} public class CloneClientBuilder {
    public class Spec {
        public String primaryAddress;
        public String backupAddress;
        public int primaryPort;
        public int backupPort;
        public String subtree;
        public long heartbeatInterval = 1000L;
    }

    private ManagedContext context;
    private Spec spec = new Spec();

    public CloneClientBuilder(ManagedContext context) {
        this.context = context;
    }

    public CloneClientBuilder withPrimaryAddress(String primaryAddress) {
        spec.primaryAddress = primaryAddress;
        return this;
    }

    public CloneClientBuilder withBackupAddress(String backupAddress) {
        spec.backupAddress = backupAddress;
        return this;
    }

    public CloneClientBuilder withPrimaryPort(int primaryPort) {
        spec.primaryPort = primaryPort;
        return this;
    }

    public CloneClientBuilder withBackupPort(int backupPort) {
        spec.backupPort = backupPort;
        return this;
    }

    public CloneClientBuilder withSubtree(String subtree) {
        spec.subtree = subtree;
        return this;
    }

    public CloneClientBuilder withHeartbeatInterval(long heartbeatInterval) {
        spec.heartbeatInterval = heartbeatInterval;
        return this;
    }

    public CloneClient build() {
        CloneClient cloneClient = new CloneClientImpl(context, spec.heartbeatInterval);
        if (spec.subtree != null) {
            cloneClient.subscribe(spec.subtree);
        }
        if (spec.primaryAddress != null) {
            cloneClient.connect(spec.primaryAddress, spec.primaryPort);
        }
        if (spec.backupAddress != null) {
            cloneClient.connect(spec.backupAddress, spec.backupPort);
        }
        
        return cloneClient;
    }
} public class CloneServerBuilder {
    public class Spec {
        public BinaryStarReactor.Mode mode;
        public String primaryAddress;
        public String backupAddress;
        public int primaryPort;
        public int backupPort;
        public int primaryBstarPort;
        public int backupBstarPort;
        public long heartbeatInterval = 1000;
    }

    private ManagedContext context;
    private Spec spec = new Spec();

    public CloneServerBuilder(ManagedContext context) {
        this.context = context;
    }

    public CloneServerBuilder withMode(BinaryStarReactor.Mode mode) {
        spec.mode = mode;
        return this;
    }

    public CloneServerBuilder withPrimaryAddress(String primaryAddress) {
        spec.primaryAddress = primaryAddress;
        return this;
    }

    public CloneServerBuilder withBackupAddress(String backupAddress) {
        spec.backupAddress = backupAddress;
        return this;
    }

    public CloneServerBuilder withPrimaryPort(int primaryPort) {
        spec.primaryPort = primaryPort;
        return this;
    }

    public CloneServerBuilder withBackupPort(int backupPort) {
        spec.backupPort = backupPort;
        return this;
    }

    public CloneServerBuilder withPrimaryBinaryStarPort(int primaryBstarPort) {
        spec.primaryBstarPort = primaryBstarPort;
        return this;
    }

    public CloneServerBuilder withBackupBinaryStarPort(int backupBstarPort) {
        spec.backupBstarPort = backupBstarPort;
        return this;
    }

    public CloneServerBuilder withHeartbeatInterval(long heartbeatInterval) {
        spec.heartbeatInterval = heartbeatInterval;
        return this;
    }

    public CloneServer build() {
        String peerAddress;
        int localPort, peerPort, localBstarPort, peerBstarPort;
        if (spec.mode == BinaryStarReactor.Mode.PRIMARY) {
            localPort = spec.primaryPort;
            peerAddress = spec.backupAddress;
            peerPort = spec.backupPort;
            localBstarPort = spec.primaryBstarPort;
            peerBstarPort = spec.backupBstarPort;
        } else {
            localPort = spec.backupPort;
            peerAddress = spec.primaryAddress;
            peerPort = spec.primaryPort;
            localBstarPort = spec.backupBstarPort;
            peerBstarPort = spec.primaryBstarPort;
        }

        CloneServerImpl cloneServer = new CloneServerImpl(context, spec.mode, peerAddress, localPort, peerPort, localBstarPort, peerBstarPort);
        cloneServer.setHeartbeatInterval(spec.heartbeatInterval);

        return cloneServer;
    }
} public class DeviceBuilder {
    public class Spec {
        public DeviceType deviceType;
        public String frontend;
        public String backend;
    }
    
    private ManagedContext context;
    private Spec spec = new Spec();

    public DeviceBuilder(ManagedContext context, DeviceType deviceType) {
        this.context = context;
        spec.deviceType = deviceType;
    }

    public DeviceBuilder withFrontendUrl(String frontend) {
        spec.frontend = frontend;
        return this;
    }

    public DeviceBuilder withBackendUrl(String backend) {
        spec.backend = backend;
        return this;
    }

    public void start() {
        switch (spec.deviceType) {
            case STREAMER:
                startStreamer();
                break;
            case FORWARDER:
                startForwarder();
                break;
            case QUEUE:
                startQueue();
                break;
        }
    }

    private void startStreamer() {
        Socket frontend = context.buildSocket(SocketType.PULL)
            .bind(spec.frontend);

        Socket backend = context.buildSocket(SocketType.PUSH)
            .bind(spec.backend);

        context.forward(frontend, backend);
    }

    private void startForwarder() {
        Socket frontend = context.buildSocket(SocketType.XSUB)
            .bind(spec.frontend);

        Socket backend = context.buildSocket(SocketType.XPUB)
            .bind(spec.backend);

        context.forward(frontend, backend);
    }

    private void startQueue() {
        Socket frontend = context.buildSocket(SocketType.ROUTER)
            .bind(spec.frontend);

        Socket backend = context.buildSocket(SocketType.DEALER)
            .bind(spec.backend);

        context.forward(frontend, backend);
    }
} public class PollableImpl implements Pollable {
    private final Socket socket;
    private final SelectableChannel channel;
    private final PollerType[] options;

    public PollableImpl(Socket socket, PollerType... options) {
        this.socket = socket;
        this.channel = null;
        this.options = options;
    }

    public PollableImpl(SelectableChannel channel, PollerType... options) {
        this.socket = null;
        this.channel = channel;
        this.options = options;
    }

    @Override
    public Socket getSocket() {
        return socket;
    }

    @Override
    public SelectableChannel getChannel() {
        return channel;
    }

    @Override
    public EnumSet<PollerType> getOptions() {
        if (options.length == 0) {
            return EnumSet.allOf(PollerType.class);
        }
        return EnumSet.copyOf(Arrays.asList(options));
    }
} public class PollerBuilder {

    private final ManagedContext context;
    private final Map<Pollable, PollListener> pollablesAndListeners = new LinkedHashMap<>();

    public PollerBuilder(ManagedContext context) {
        this.context = context;
    }

    public PollerBuilder withPollable(Pollable pollable, PollListener listener) {
        pollablesAndListeners.put(pollable, listener);
        return this;
    }

    /*
     * Socket Pollables.
     */

    public PollerBuilder withInPollable(Socket socket, PollListener listener) {
        return withPollable(context.newPollable(socket, PollerType.POLL_IN), listener);
    }

    public PollerBuilder withOutPollable(Socket socket, PollListener listener) {
        return withPollable(context.newPollable(socket, PollerType.POLL_OUT), listener);
    }

    public PollerBuilder withErrorPollable(Socket socket, PollListener listener) {
        return withPollable(context.newPollable(socket, PollerType.POLL_ERROR), listener);
    }

    public PollerBuilder withInOutPollable(Socket socket, PollListener listener) {
        return withPollable(context.newPollable(socket, PollerType.POLL_IN, PollerType.POLL_OUT), listener);
    }

    public PollerBuilder withAllPollable(Socket socket, PollListener listener) {
        return withPollable(context.newPollable(socket, PollerType.POLL_IN, PollerType.POLL_OUT, PollerType.POLL_ERROR), listener);
    }

    /*
     * SelectableChannel Pollables.
     */

    public PollerBuilder withInPollable(SelectableChannel channel, PollListener listener) {
        return withPollable(context.newPollable(channel, PollerType.POLL_IN), listener);
    }

    public PollerBuilder withOutPollable(SelectableChannel channel, PollListener listener) {
        return withPollable(context.newPollable(channel, PollerType.POLL_OUT), listener);
    }

    public PollerBuilder withErrorPollable(SelectableChannel channel, PollListener listener) {
        return withPollable(context.newPollable(channel, PollerType.POLL_ERROR), listener);
    }

    public PollerBuilder withInOutPollable(SelectableChannel channel, PollListener listener) {
        return withPollable(context.newPollable(channel, PollerType.POLL_IN, PollerType.POLL_OUT), listener);
    }

    public PollerBuilder withAllPollable(SelectableChannel channel, PollListener listener) {
        return withPollable(context.newPollable(channel, PollerType.POLL_IN, PollerType.POLL_OUT, PollerType.POLL_ERROR), listener);
    }

    @Deprecated
    public Poller create() {
        return build();
    }

    public Poller build() {
        return new PollerImpl(context, pollablesAndListeners);
    }
} public class DealerSocketBuilder extends SocketBuilder {
    public DealerSocketBuilder(ManagedContext context) {
        super(context, SocketType.DEALER);
    }

} public class PairSocketBuilder extends SocketBuilder {

    public PairSocketBuilder(ManagedContext context) {
        super(context, SocketType.PAIR);
    }

} public class PubSocketBuilder extends SocketBuilder {

    public PubSocketBuilder(ManagedContext context) {
        super(context, SocketType.PUB);
    }

} public class PullSocketBuilder extends SocketBuilder {

    public PullSocketBuilder(ManagedContext context) {
        super(context, SocketType.PULL);
    }

} public class PushSocketBuilder extends SocketBuilder {

    public PushSocketBuilder(ManagedContext context) {
        super(context, SocketType.PUSH);
    }

} public class RepSocketBuilder extends SocketBuilder {
    public RepSocketBuilder(ManagedContext managedContext) {
        super(managedContext, SocketType.REP);
    }

} public class ReqSocketBuilder extends SocketBuilder {
    public ReqSocketBuilder(ManagedContext context) {
        super(context, SocketType.REQ);
    }


} public class RouterSocketBuilder extends SocketBuilder implements Routable {
    private boolean routerMandatory = false;

    public RouterSocketBuilder(ManagedContext managedContext) {
        super(managedContext, SocketType.ROUTER);
    }

    @Override
    public Socket connect(String url, String... additionalUrls) {
        ZMQ.Socket socket = createConnectableSocketWithStandardSettings();
        socket.setRouterMandatory(routerMandatory);
        connect(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }

    @Override
    public Socket bind(String url, String... additionalUrls) {
        ZMQ.Socket socket = createBindableSocketWithStandardSettings();
        socket.setRouterMandatory(routerMandatory);
        bind(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }

    @Override
    public SocketBuilder withRouterMandatory() {
        routerMandatory = true;
        return this;
    }
} public class SocketBuilder implements Bindable, Connectable {
    private static final String[] EMPTY_ADDITIONAL_URLS = new String[0];

    protected ManagedContext context;
    protected SocketSpec socketSpec;

    public class SocketSpec {
        public long swapSize;
        public int linger;
        public int sendHighwatermark;
        public int receiveHighWatermark;
        public int receiveTimeout = -1;
        public int sendTimeout = -1;
        public SocketType socketType;
        public TransportType transportType;
        public byte[] identity;
        public Backgroundable backgroundable;
    }

    public SocketBuilder(ManagedContext context, SocketType socketType) {
        this.socketSpec = new SocketSpec();
        this.socketSpec.socketType = socketType;
        this.context = context;
    }

    /**
     * Returns the underlying socket type
     * 
     * @return socket type
     */
    public SocketType getSocketType() {
        return socketSpec.socketType;
    }

    /**
     * Set the linger period for the specified socket. The linger period determines how long pending which have yet to
     * sent to a peer shall linger in memory after a socket is closed.
     * 
     * @param lingerMS the linger period in millis
     * @return builder object
     * @deprecated The linger option has only integer range, use {@link SocketBuilder#withLinger(int)} instead.
     */
    @Deprecated
    public SocketBuilder withLinger(long lingerMS) {
        getSocketSpec().linger = Long.valueOf(lingerMS).intValue();
        return this;
    }

    /**
     * Set the linger period for the specified socket. The linger period determines how long pending which have yet to
     * sent to a peer shall linger in memory after a socket is closed.
     *
     * @param lingerMS the linger period in millis
     * @return builder object
     */
    public SocketBuilder withLinger(int lingerMS) {
        getSocketSpec().linger = lingerMS;
        return this;
    }

    /**
     * Returns the linger period in millis
     * 
     * @return linger period
     */
    public int getLinger() {
        return getSocketSpec().linger;
    }

    /**
     * Set the socket identity of the specified socket. Socket identity determines if existing 0MQ infrastructure
     * (message queues, forwarding devices) shall be identified with a specific application and persist across multiple
     * runs of the application.
     * 
     * If the socket has no identity, each run of the application is completely independent of other runs.
     * 
     * The identity can only be between 0 and 256 bytes long exclusive.
     * 
     * @param identity the identity
     * @return builder object
     */
    public SocketBuilder withIdentity(byte[] identity) {
        getSocketSpec().identity = identity;
        return this;
    }

    /**
     * Return the identity of the socket
     * 
     * @return the identity
     */
    public byte[] getIdentity() {
        return getSocketSpec().identity;
    }

    /**
     * The swap option shall set the disk offload (swap) size for the specified socket. A socket which has a positive
     * value may exceed it's high water mark; in the case of outstanding messages, they shall be offloaded to storage on
     * disk.
     * 
     * @param swapSize swap in bytes
     * @return builder object
     */
    public SocketBuilder withSwap(long swapSize) {
        getSocketSpec().swapSize = swapSize;
        return this;
    }

    /**
     * Return the swap size in bytes
     * 
     * @return the swap size
     */
    public long getSwap() {
        return getSocketSpec().swapSize;
    }

    /**
     * Set the send high watermark.
     * 
     * @param sendHWM The send high watermark
     * @return builder object
     * @deprecated This option uses integer range, use {@link SocketBuilder#withSendHighWatermark(int)} instead.
     */
    @Deprecated
    public SocketBuilder withSendHighWatermark(long sendHWM) {
        getSocketSpec().sendHighwatermark = Long.valueOf(sendHWM).intValue();
        return this;
    }

    /**
     * Set the send high watermark.
     *
     * @param sendHWM The send high watermark
     * @return builder object
     */
    public SocketBuilder withSendHighWatermark(int sendHWM) {
        getSocketSpec().sendHighwatermark = sendHWM;
        return this;
    }

    /**
     * Get the send high watermark.
     * 
     * @return send high watermark
     */
    public int getSendHighWaterMark() {
        return getSocketSpec().sendHighwatermark;
    }

    /**
     * The RECVHWM option shall set the high water mark (HWM) for inbound messages on the specified socket. The HWM is a
     * hard limit on the maximum number of outstanding 0MQ shall queue in memory for any single peer that the specified
     * socket is communicating with.
     * 
     * If this limit has been reached the socket shall enter an exceptional state and depending on the socket type, 0MQ
     * shall take appropriate action such as blocking or dropping sent messages. Refer to the individual socket
     * descriptions in <a href="http://api.zeromq.org/3-2:zmq-socket">zmq_socket(3)</a> for details on the exact action
     * taken for each socket type.
     * 
     * @param receiveHWM recv high water mark
     * @return builder object
     * @deprecated This option uses integer range, use {@link #withReceiveHighWatermark(int)} instead.
     */
    @Deprecated
    public SocketBuilder withReceiveHighWatermark(long receiveHWM) {
        getSocketSpec().receiveHighWatermark = Long.valueOf(receiveHWM).intValue();
        return this;
    }

    /**
     * The RECVHWM option shall set the high water mark (HWM) for inbound messages on the specified socket. The HWM is a
     * hard limit on the maximum number of outstanding 0MQ shall queue in memory for any single peer that the specified
     * socket is communicating with.
     *
     * If this limit has been reached the socket shall enter an exceptional state and depending on the socket type, 0MQ
     * shall take appropriate action such as blocking or dropping sent messages. Refer to the individual socket
     * descriptions in <a href="http://api.zeromq.org/3-2:zmq-socket">zmq_socket(3)</a> for details on the exact action
     * taken for each socket type.
     *
     * @param receiveHWM recv high water mark
     * @return builder object
     */
    public SocketBuilder withReceiveHighWatermark(int receiveHWM) {
        getSocketSpec().receiveHighWatermark = receiveHWM;
        return this;
    }

    /**
     * Returns the receive high water mark.
     * 
     * @return receive high water mark
     */
    public int getReceiveHighWaterMark() {
        return getSocketSpec().receiveHighWatermark;
    }

    public SocketSpec getSocketSpec() {
        return socketSpec;
    }

    public SocketBuilder withReceiveTimeout(int receiveTimeout) {
        getSocketSpec().receiveTimeout = receiveTimeout;
        return this;
    }

    public SocketBuilder withSendTimeout(int sendTimeout) {
        getSocketSpec().sendTimeout = sendTimeout;
        return this;
    }

    public SocketBuilder withBackgroundable(Backgroundable backgroundable) {
        getSocketSpec().backgroundable = backgroundable;
        return this;
    }

    //todo should these be here & removed from the subclasses?  They appear to all be the same implementations.
    /**
     * {@inheritDoc}
     */
    @Override
    public Socket connect(String url) {
        return connect(url, EMPTY_ADDITIONAL_URLS);
    }

    @Override
    public Socket connect(String url, String... additionalUrls) {
        ZMQ.Socket socket = createConnectableSocketWithStandardSettings();
        connect(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }

    protected void connect(ZMQ.Socket socket, String url, String[] additionalUrls) {
        socket.connect(url);
        for (String s : additionalUrls) {
            socket.connect(s);
        }
    }

    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;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Socket bind(String url) {
        return bind(url, EMPTY_ADDITIONAL_URLS);
    }

    @Override
    public Socket bind(String url, String... additionalUrls) {
        ZMQ.Socket socket = createBindableSocketWithStandardSettings();
        bind(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }

    protected void bind(ZMQ.Socket socket, String url, String[] additionalUrls) {
        socket.bind(url);
        for (String s : additionalUrls) {
            socket.bind(s);
        }
    }

    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;
    }

    protected Socket newManagedSocket(ZMQ.Socket socket) {
        ManagedSocket managedSocket = new ManagedSocket(context, socket);
        if (getSocketSpec().backgroundable != null) {
            context.fork(managedSocket, getSocketSpec().backgroundable);
        }
        return managedSocket;
    }

    /**
     * Coerce the SocketBuilder to be Subscribable.
     * 
     * @return This builder object as a Subscribable
     */
    public Subscribable asSubscribable() {
        return (Subscribable) this;
    }

    /**
     * Coerce the SocketBuilder to be Routable.
     * 
     * @return This builder object as a Routable
     */
    public Routable asRoutable() {
        return (RouterSocketBuilder) this;
    }
} public class SubSocketBuilder extends SocketBuilder implements Subscribable {
    private List<byte[]> subscriptions = new ArrayList<>();

    public SubSocketBuilder(ManagedContext context) {
        super(context, SocketType.SUB);
    }

    @Override
    public SubSocketBuilder subscribe(byte[] data) {
        byte[] subscription = new byte[data.length];
        System.arraycopy(data, 0, subscription, 0, data.length);
        subscriptions.add(subscription);
        return this;
    }

    @Override
    public SubSocketBuilder subscribeAll() {
        return subscribe(new byte[0]);
    }

    @Override
    public Socket connect(String url, String... additionalUrls) {
        if (subscriptions.isEmpty()) {
            throw new IllegalStateException("You must have a SUB socket subscribe to something before you can connect it.");
        }
        ZMQ.Socket socket = createConnectableSocketWithStandardSettings();
        for (byte[] subscription : subscriptions) {
            socket.subscribe(subscription);
        }
        connect(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }
    
    @Override
    public Socket bind(String url, String... additionalUrls) {
        if (subscriptions.isEmpty()) {
            throw new IllegalStateException("You must have a SUB socket subscribe to something before you can bind it.");
        }
        ZMQ.Socket socket = createConnectableSocketWithStandardSettings();
        for (byte[] subscription : subscriptions) {
            socket.subscribe(subscription);
        }
        bind(socket, url, additionalUrls);
        return newManagedSocket(socket);
    }

} public class XPubSocketBuilder extends SocketBuilder {

    public XPubSocketBuilder(ManagedContext context) {
        super(context, SocketType.XPUB);
    }

} public class XSubSocketBuilder extends SocketBuilder {

    public XSubSocketBuilder(ManagedContext context) {
        super(context, SocketType.XSUB);
    }

}
import java.nio.channels.SelectableChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;
import org.zeromq.ZMQException;
import org.zeromq.api.Backgroundable;
import org.zeromq.api.Context;
import org.zeromq.api.DeviceType;
import org.zeromq.api.Pollable;
import org.zeromq.api.PollerType;
import org.zeromq.api.Socket;
import org.zeromq.api.SocketType;
import org.zeromq.api.exception.ZMQExceptions;
import org.zeromq.jzmq.beacon.BeaconReactorBuilder;
import org.zeromq.jzmq.bstar.BinaryStarReactorBuilder;
import org.zeromq.jzmq.bstar.BinaryStarSocketBuilder;
import org.zeromq.jzmq.clone.CloneClientBuilder;
import org.zeromq.jzmq.clone.CloneServerBuilder;
import org.zeromq.jzmq.device.DeviceBuilder;
import org.zeromq.jzmq.poll.PollableImpl;
import org.zeromq.jzmq.poll.PollerBuilder;
import org.zeromq.jzmq.reactor.ReactorBuilder;
import org.zeromq.jzmq.sockets.DealerSocketBuilder;
import org.zeromq.jzmq.sockets.PairSocketBuilder;
import org.zeromq.jzmq.sockets.PubSocketBuilder;
import org.zeromq.jzmq.sockets.PullSocketBuilder;
import org.zeromq.jzmq.sockets.PushSocketBuilder;
import org.zeromq.jzmq.sockets.RepSocketBuilder;
import org.zeromq.jzmq.sockets.ReqSocketBuilder;
import org.zeromq.jzmq.sockets.RouterSocketBuilder;
import org.zeromq.jzmq.sockets.SocketBuilder;
import org.zeromq.jzmq.sockets.SubSocketBuilder;
import org.zeromq.jzmq.sockets.XPubSocketBuilder;
import org.zeromq.jzmq.sockets.XSubSocketBuilder;
package org.zeromq.jzmq;



/**
 * Manage JZMQ Context
 */
public class ManagedContext implements Context {
    private static final Logger log = LoggerFactory.getLogger(ManagedContext.class);

    private final AtomicBoolean closed = new AtomicBoolean(false);

    private boolean termContext;
    private final ZMQ.Context context;
    private final Set<Socket> sockets;
    private final List<Backgroundable> backgroundables;

    public ManagedContext() {
        this(ZMQ.context(1));
    }

    public ManagedContext(int ioThreads) {
        this(ZMQ.context(ioThreads));
    }

    public ManagedContext(ZMQ.Context context) {
        this(context, true);
    }

    public ManagedContext(ZMQ.Context context, boolean termContext) {
        if (context == null) {
            throw new IllegalArgumentException("Context cannot be null");
        }
        this.sockets = new CopyOnWriteArraySet<>();
        this.backgroundables = new ArrayList<>();
        this.context = context;
        this.termContext = termContext;
    }

    public ZMQ.Context getZMQContext() {
        return context;
    }

    // Do people actually need this?
    public Collection<Socket> getSockets() {
        return Collections.unmodifiableCollection(sockets);
    }

    public void destroySocket(Socket socket) {
        if (sockets.contains(socket)) {
            try {
                socket.close();
            } catch (Exception ignore) {
                log.warn("Exception caught while closing underlying socket.", ignore);
            }
            log.debug("closed socket");
        }
    }

    @Override
    public ManagedContext shadow() {
        return new ManagedContext(context, false);
    }

    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            for (Backgroundable b : backgroundables) {
                b.onClose();
            }
            for (Socket s : sockets) {
                destroySocket(s);
            }
            sockets.clear();
            if (termContext) {
                context.term();
            }
            log.debug("closed context");
        }
    }

    @Override
    public void terminate() {
        if (termContext) {
            // Terminate context on another thread
            TermThread thread = new TermThread(this);
            thread.start();
            try {
                // Wait until thread has started, and a bit more
                thread.await();
                Thread.sleep(15);
            } catch (InterruptedException ignored) {
            }

            // Don't double-terminate context
            termContext = false;
        }
    }

    // I'd really like this to be private but I don't want all the builders in here
    // If we only deal with the Context interface, callers won't see this
    void addSocket(Socket socket) {
        sockets.add(socket);
    }

    @Override
    public SocketBuilder buildSocket(SocketType type) {
        switch (type) {
            case PULL:
                return new PullSocketBuilder(this);
            case PUSH:
                return new PushSocketBuilder(this);
            case PUB:
                return new PubSocketBuilder(this);
            case SUB:
                return new SubSocketBuilder(this);
            case XPUB:
                return new XPubSocketBuilder(this);
            case XSUB:
                return new XSubSocketBuilder(this);
            case REP:
                return new RepSocketBuilder(this);
            case REQ:
                return new ReqSocketBuilder(this);
            case ROUTER:
                return new RouterSocketBuilder(this);
            case DEALER:
                return new DealerSocketBuilder(this);