@Test public void testGetProducerMessageFuturePairs() throws Exception { String topic = "topic"; long bornTime = System.currentTimeMillis(); SendMessageCommand cmd = new SendMessageCommand("topic", 10); ProducerMessage<String> pmsg1 = createProducerMessage(topic, "body1", "key1", 10, "pKey1", bornTime, true, true, Arrays.asList(new Pair<String, String>("a1", "A1")), Arrays.asList(new Pair<String, String>("b1", "B1")), Arrays.asList(new Pair<String, String>("c1", "C1"))); SettableFuture<SendResult> future1 = SettableFuture.create(); cmd.addMessage(pmsg1, future1); ProducerMessage<String> pmsg2 = createProducerMessage(topic, "body2", "key2", 10, "pKey2", bornTime, true, true, Arrays.asList(new Pair<String, String>("a2", "A2")), Arrays.asList(new Pair<String, String>("b2", "B2")), Arrays.asList(new Pair<String, String>("c2", "C2"))); SettableFuture<SendResult> future2 = SettableFuture.create(); cmd.addMessage(pmsg2, future2); ProducerMessage<String> pmsg3 = createProducerMessage(topic, "body3", "key3", 10, "pKey3", bornTime, false, true, Arrays.asList(new Pair<String, String>("a3", "A3")), Arrays.asList(new Pair<String, String>("b3", "B3")), Arrays.asList(new Pair<String, String>("c3", "C3"))); SettableFuture<SendResult> future3 = SettableFuture.create(); cmd.addMessage(pmsg3, future3); ProducerMessage<String> pmsg4 = createProducerMessage(topic, "body4", "key4", 10, "pKey4", bornTime, false, true, Arrays.asList(new Pair<String, String>("a4", "A4")), Arrays.asList(new Pair<String, String>("b4", "B4")), Arrays.asList(new Pair<String, String>("c4", "C4"))); SettableFuture<SendResult> future4 = SettableFuture.create(); cmd.addMessage(pmsg4, future4); List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = cmd.getProducerMessageFuturePairs(); assertEquals(4, pairs.size()); assertSame(pmsg1, pairs.get(0).getKey()); assertSame(future1, pairs.get(0).getValue()); assertSame(pmsg2, pairs.get(1).getKey()); assertSame(future2, pairs.get(1).getValue()); assertSame(pmsg3, pairs.get(2).getKey()); assertSame(future3, pairs.get(2).getValue()); assertSame(pmsg4, pairs.get(3).getKey()); assertSame(future4, pairs.get(3).getValue()); }
public List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs() { List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = new LinkedList<Pair<ProducerMessage<?>, SettableFuture<SendResult>>>(); Collection<List<ProducerMessage<?>>> msgsList = getProducerMessages(); for (List<ProducerMessage<?>> msgs : msgsList) { for (ProducerMessage<?> msg : msgs) { SettableFuture<SendResult> future = m_futures.get(msg.getMsgSeqNo()); if (future != null) { pairs.add(new Pair<ProducerMessage<?>, SettableFuture<SendResult>>(msg, future)); } } } return pairs; }
SendMessageCommand extends AbstractCommand { public List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs() { List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = new LinkedList<Pair<ProducerMessage<?>, SettableFuture<SendResult>>>(); Collection<List<ProducerMessage<?>>> msgsList = getProducerMessages(); for (List<ProducerMessage<?>> msgs : msgsList) { for (ProducerMessage<?> msg : msgs) { SettableFuture<SendResult> future = m_futures.get(msg.getMsgSeqNo()); if (future != null) { pairs.add(new Pair<ProducerMessage<?>, SettableFuture<SendResult>>(msg, future)); } } } return pairs; } }
SendMessageCommand extends AbstractCommand { public List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs() { List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = new LinkedList<Pair<ProducerMessage<?>, SettableFuture<SendResult>>>(); Collection<List<ProducerMessage<?>>> msgsList = getProducerMessages(); for (List<ProducerMessage<?>> msgs : msgsList) { for (ProducerMessage<?> msg : msgs) { SettableFuture<SendResult> future = m_futures.get(msg.getMsgSeqNo()); if (future != null) { pairs.add(new Pair<ProducerMessage<?>, SettableFuture<SendResult>>(msg, future)); } } } return pairs; } SendMessageCommand();  SendMessageCommand(String topic, int partition); }
SendMessageCommand extends AbstractCommand { public List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs() { List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = new LinkedList<Pair<ProducerMessage<?>, SettableFuture<SendResult>>>(); Collection<List<ProducerMessage<?>>> msgsList = getProducerMessages(); for (List<ProducerMessage<?>> msgs : msgsList) { for (ProducerMessage<?> msg : msgs) { SettableFuture<SendResult> future = m_futures.get(msg.getMsgSeqNo()); if (future != null) { pairs.add(new Pair<ProducerMessage<?>, SettableFuture<SendResult>>(msg, future)); } } } return pairs; } SendMessageCommand();  SendMessageCommand(String topic, int partition); ConcurrentMap<Integer, List<ProducerMessage<?>>> getMsgs(); String getTopic(); int getPartition(); void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future); Map<Integer, MessageBatchWithRawData> getMessageRawDataBatches(); int getMessageCount(); void onResultReceived(SendMessageResultCommand result); @Override void parse0(ByteBuf buf); @Override void toBytes0(ByteBuf buf); Collection<List<ProducerMessage<?>>> getProducerMessages(); List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs(); Map<Integer, SettableFuture<SendResult>> getFutures(); }
SendMessageCommand extends AbstractCommand { public List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs() { List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> pairs = new LinkedList<Pair<ProducerMessage<?>, SettableFuture<SendResult>>>(); Collection<List<ProducerMessage<?>>> msgsList = getProducerMessages(); for (List<ProducerMessage<?>> msgs : msgsList) { for (ProducerMessage<?> msg : msgs) { SettableFuture<SendResult> future = m_futures.get(msg.getMsgSeqNo()); if (future != null) { pairs.add(new Pair<ProducerMessage<?>, SettableFuture<SendResult>>(msg, future)); } } } return pairs; } SendMessageCommand();  SendMessageCommand(String topic, int partition); ConcurrentMap<Integer, List<ProducerMessage<?>>> getMsgs(); String getTopic(); int getPartition(); void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future); Map<Integer, MessageBatchWithRawData> getMessageRawDataBatches(); int getMessageCount(); void onResultReceived(SendMessageResultCommand result); @Override void parse0(ByteBuf buf); @Override void toBytes0(ByteBuf buf); Collection<List<ProducerMessage<?>>> getProducerMessages(); List<Pair<ProducerMessage<?>, SettableFuture<SendResult>>> getProducerMessageFuturePairs(); Map<Integer, SettableFuture<SendResult>> getFutures();  }