ZeromqConsumer extends DefaultConsumer { @Override protected void doStart() throws Exception { super.doStart(); executor = endpoint.getCamelContext().getExecutorServiceManager().newFixedThreadPool(this, endpoint.getEndpointUri(), 1); listener = new Listener(endpoint, processor, socketFactory, contextFactory); executor.submit(listener); } ZeromqConsumer(ZeromqEndpoint endpoint, Processor processor, ContextFactory contextFactory, SocketFactory socketFactory); ContextFactory getContextFactory(); SocketFactory getSocketFactory(); @Override void resume(); @Override void suspend();  }
@Test public void connect() throws Exception { ExecutorService executor = Executors.newFixedThreadPool(1); when(manager.newFixedThreadPool(any(ZeromqConsumer.class), anyString(), anyInt())).thenReturn(executor); consumer.doStart(); Thread.sleep(50); verify(socket).connect(endpoint.getSocketAddress()); }
@Test public void bind() throws Exception { ExecutorService executor = Executors.newFixedThreadPool(1); when(manager.newFixedThreadPool(any(ZeromqConsumer.class), anyString(), anyInt())).thenReturn(executor); when(endpoint.getMode()).thenReturn("BIND"); consumer.doStart(); Thread.sleep(50); verify(socket).bind(endpoint.getSocketAddress()); }
@Test public void listenerIsSubmittedOnStart() throws Exception { consumer.doStart(); verify(executor).submit(any(Runnable.class)); }
@Test @Ignore public void connect() throws Exception { ExecutorService executor = Executors.newFixedThreadPool(1); when(manager.newFixedThreadPool(any(ZeromqConsumer.class), anyString(), anyInt())).thenReturn(executor); consumer.doStart(); Thread.sleep(50); verify(socket).connect(endpoint.getSocketAddress()); }