@Test(expected = IllegalArgumentException.class) public void testTopicNotMatch() throws Exception { String topic = "topic"; long bornTime = System.currentTimeMillis(); SendMessageCommand cmd = new SendMessageCommand("topic", 10); ProducerMessage<String> pmsg1 = createProducerMessage(topic + "1111", "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); }
public void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future) { validate(msg); int msgSeqNo = m_msgCounter.getAndIncrement(); msg.setMsgSeqNo(msgSeqNo); if (msg.isPriority()) { if (!m_msgs.containsKey(0)) { m_msgs.putIfAbsent(0, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(0).add(msg); } else { if (!m_msgs.containsKey(1)) { m_msgs.putIfAbsent(1, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(1).add(msg); } m_futures.put(msgSeqNo, future); }
SendMessageCommand extends AbstractCommand { public void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future) { validate(msg); int msgSeqNo = m_msgCounter.getAndIncrement(); msg.setMsgSeqNo(msgSeqNo); if (msg.isPriority()) { if (!m_msgs.containsKey(0)) { m_msgs.putIfAbsent(0, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(0).add(msg); } else { if (!m_msgs.containsKey(1)) { m_msgs.putIfAbsent(1, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(1).add(msg); } m_futures.put(msgSeqNo, future); } }
SendMessageCommand extends AbstractCommand { public void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future) { validate(msg); int msgSeqNo = m_msgCounter.getAndIncrement(); msg.setMsgSeqNo(msgSeqNo); if (msg.isPriority()) { if (!m_msgs.containsKey(0)) { m_msgs.putIfAbsent(0, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(0).add(msg); } else { if (!m_msgs.containsKey(1)) { m_msgs.putIfAbsent(1, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(1).add(msg); } m_futures.put(msgSeqNo, future); } SendMessageCommand();  SendMessageCommand(String topic, int partition); }
SendMessageCommand extends AbstractCommand { public void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future) { validate(msg); int msgSeqNo = m_msgCounter.getAndIncrement(); msg.setMsgSeqNo(msgSeqNo); if (msg.isPriority()) { if (!m_msgs.containsKey(0)) { m_msgs.putIfAbsent(0, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(0).add(msg); } else { if (!m_msgs.containsKey(1)) { m_msgs.putIfAbsent(1, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(1).add(msg); } m_futures.put(msgSeqNo, future); } 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 void addMessage(ProducerMessage<?> msg, SettableFuture<SendResult> future) { validate(msg); int msgSeqNo = m_msgCounter.getAndIncrement(); msg.setMsgSeqNo(msgSeqNo); if (msg.isPriority()) { if (!m_msgs.containsKey(0)) { m_msgs.putIfAbsent(0, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(0).add(msg); } else { if (!m_msgs.containsKey(1)) { m_msgs.putIfAbsent(1, new LinkedList<ProducerMessage<?>>()); } m_msgs.get(1).add(msg); } m_futures.put(msgSeqNo, future); } 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();  }