@Test public void testEMQSpout() throws InterruptedException { int sent = 0; MessageService.Iface messageClient = new EMQClientFactory(QueueUtil.getCredential()).newMessageClient(EMQTestConfig.EMQ_ENDPOINT); for (int i = 0; i < 100; i++) { SendMessageRequest request = new SendMessageRequest(queueName, i + ""); try { messageClient.sendMessage(request); sent++; } catch (TException e) { } } executor.exec(); System.out.println("Emited: " + executor.getEmitCount() + "\nAcked: " + executor.getAckCount() + "\nFailed: " + executor.getFailCount()); Assert.assertTrue("Emited messages number of EMQSpout should be more than messages sent", executor.getEmitCount() >= sent); }
public EMQSpout(EMQConfig emqConfig) { this.emqConfig = emqConfig; }
EMQSpout extends BaseRichSpout { public EMQSpout(EMQConfig emqConfig) { this.emqConfig = emqConfig; } }
EMQSpout extends BaseRichSpout { public EMQSpout(EMQConfig emqConfig) { this.emqConfig = emqConfig; } EMQSpout(EMQConfig emqConfig); }
EMQSpout extends BaseRichSpout { public EMQSpout(EMQConfig emqConfig) { this.emqConfig = emqConfig; } EMQSpout(EMQConfig emqConfig); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector); @Override void close(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); }
EMQSpout extends BaseRichSpout { public EMQSpout(EMQConfig emqConfig) { this.emqConfig = emqConfig; } EMQSpout(EMQConfig emqConfig); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector); @Override void close(); @Override void nextTuple(); @Override void ack(Object msgId); @Override void fail(Object msgId); static final Logger LOG; }