ZeromqProducer extends DefaultProducer { @Override public void start() throws Exception { this.context = contextFactory.createContext(1); this.socket = socketFactory.createProducerSocket(context, endpoint.getSocketType()); this.topics = endpoint.getTopics() == null ? null : endpoint.getTopics().split(","); String addr = endpoint.getSocketAddress(); if (endpoint.getMode() == null || endpoint.getMode().equals("BIND")) { LOGGER.info("Binding client to [{}]", addr); socket.bind(addr); LOGGER.info("Bound OK"); } else { LOGGER.info("Connecting client to [{}]", addr); socket.connect(addr); LOGGER.info("Connected OK"); } } ZeromqProducer(ZeromqEndpoint endpoint, SocketFactory socketFactory, ContextFactory contextFactory); ContextFactory getContextFactory(); SocketFactory getSocketFactory(); @Override boolean isSingleton(); @Override void process(Exchange arg0); void setShutdownWait(int shutdownWait); @Override void start(); @Override void stop();  }
@Test public void bind() throws Exception { producer.start(); verify(socket).bind(endpoint.getSocketAddress()); }
@Test public void connect() throws Exception { when(endpoint.getMode()).thenReturn("CONNECT"); producer.start(); verify(socket).connect(endpoint.getSocketAddress()); }