AkkaSocketFactory implements SocketFactory { @Override public Socket createProducerSocket(Context context, ZeromqSocketType socketType) { LOGGER.debug("Creating producer socket [{}]", socketType); Socket socket; switch (socketType) { case DEALER: socket = context.socket(ZMQ.DEALER); break; case PUBLISH: socket = context.socket(ZMQ.PUB); break; case PUSH: socket = context.socket(ZMQ.PUSH); break; default: throw new ZeromqException("Unsupported socket type for producer: " + socketType); } applySocketOptions(socket); return socket; } AkkaSocketFactory(long highWaterMark, long linger); @Override Socket createConsumerSocket(Context context, ZeromqSocketType socketType); @Override Socket createProducerSocket(Context context, ZeromqSocketType socketType);  }
@Test(expected = ZeromqException.class) public void pullIsConsumer() { new AkkaSocketFactory(0, 0).createProducerSocket(null, ZeromqSocketType.PULL); }
@Test(expected = ZeromqException.class) public void subscribeIsConsumer() { new AkkaSocketFactory(0, 0).createProducerSocket(null, ZeromqSocketType.SUBSCRIBE); }