Listener implements Runnable { void connect() { context = akkaContextFactory.createContext(1); socket = akkaSocketFactory.createConsumerSocket(context, endpoint.getSocketType()); String addr = endpoint.getSocketAddress(); if (endpoint.getMode() == null || endpoint.getMode().equals("CONNECT")) { LOGGER.info("Connecting to server [{}]", addr); socket.connect(addr); LOGGER.info("Connected OK"); } else { LOGGER.info("Binding to server [{}]", addr); socket.bind(addr); LOGGER.info("Bound OK"); } if (endpoint.getSocketType() == ZeromqSocketType.SUBSCRIBE) { subscribe(); } } Listener(ZeromqEndpoint endpoint, Processor processor, SocketFactory akkaSocketFactory, ContextFactory akkaContextFactory); @Override void run(); void setCallback(AsyncCallback callback);  }
@Test public void socketConnectedWithAddress() { listener.connect(); verify(socket).connect("tcp: }
@Test public void socketCreatedWithCorrectContext() { listener.connect(); verify(akkaSocketFactory).createConsumerSocket(context, ZeromqSocketType.SUBSCRIBE); }