ZeromqEndpoint extends DefaultEndpoint { @Override public ZeromqConsumer createConsumer(Processor processor) throws Exception { if (socketType == null) { throw new ZeromqException("Must specify socket type as a parameter, eg socketType=SUBSCRIBE"); } return new ZeromqConsumer(this, processor, contextFactory, socketFactory); } ZeromqEndpoint(String endpointUri, String remaining, ZeromqComponent component); @Override ZeromqConsumer createConsumer(Processor processor); @Override ZeromqProducer createProducer(); @Override boolean isSingleton(); long getHighWaterMark(); String getHostname(); long getLinger(); @SuppressWarnings("rawtypes") Class getMessageConvertor(); int getPort(); String getProtocol(); ZeromqSocketType getSocketType(); String getTopics(); String getMode(); boolean isAsyncConsumer(); boolean isMessageIdEnabled(); void setAsyncConsumer(boolean asyncConsumer); void setContextFactory(ContextFactory contextFactory); void setHighWaterMark(long highWaterMark); void setLinger(long linger); void setMessageConvertor(Class<?> messageConvertor); void setMessageIdEnabled(boolean messageIdEnabled); void setSocketFactory(SocketFactory socketFactory); void setSocketType(ZeromqSocketType socketType); void setTopics(String topics); void setMode(String mode);  }
@Test public void createConsumer() throws Exception { ZeromqEndpoint endpoint = new ZeromqEndpoint("zeromq:tcp: endpoint.setSocketType(ZeromqSocketType.PUSH); endpoint.createConsumer(new Processor() { @Override public void process(Exchange exchange) throws Exception { } }); }
@Test(expected = ZeromqException.class) public void topicsErrorWithoutSubscribeType() throws Exception { ZeromqEndpoint endpoint = new ZeromqEndpoint("zeromq:tcp: new ZeromqComponent()); endpoint.createConsumer(new Processor() { @Override public void process(Exchange exchange) throws Exception { } }); }