@Test public void testExecuteDefault() throws SubCommandException { PrintStream out = System.out; ByteArrayOutputStream bos = new ByteArrayOutputStream(); System.setOut(new PrintStream(bos)); Options options = ServerUtil.buildCommandlineOptions(new Options()); String[] subargs = new String[] {"-t mytopic", "-n localhost:9876"}; CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin " + consumeMessageCommand.commandName(), subargs, consumeMessageCommand.buildCommandlineOptions(options), new PosixParser()); consumeMessageCommand.execute(commandLine, options, null); System.setOut(out); String s = new String(bos.toByteArray()); Assert.assertTrue(s.contains("Consume ok")); }
private void executeDefault(long timeValueBegin, long timeValueEnd) { try { Set<MessageQueue> mqs = defaultMQPullConsumer.fetchSubscribeMessageQueues(topic); long countLeft = messageCount; for (MessageQueue mq : mqs) { if (countLeft == 0) { return; } long minOffset = defaultMQPullConsumer.minOffset(mq); long maxOffset = defaultMQPullConsumer.maxOffset(mq); if (timeValueBegin > 0) { minOffset = defaultMQPullConsumer.searchOffset(mq, timeValueBegin); } if (timeValueEnd > 0) { maxOffset = defaultMQPullConsumer.searchOffset(mq, timeValueEnd); } if (maxOffset - minOffset > countLeft) { System.out.printf("The older %d message of the %d queue will be provided\n", countLeft, mq.getQueueId()); maxOffset = minOffset + countLeft - 1; countLeft = 0; } else { countLeft = countLeft - (maxOffset - minOffset) - 1; } pullMessageByQueue(mq, minOffset, maxOffset); } } catch (Exception e) { e.printStackTrace(); } }
ConsumeMessageCommand implements SubCommand { private void executeDefault(long timeValueBegin, long timeValueEnd) { try { Set<MessageQueue> mqs = defaultMQPullConsumer.fetchSubscribeMessageQueues(topic); long countLeft = messageCount; for (MessageQueue mq : mqs) { if (countLeft == 0) { return; } long minOffset = defaultMQPullConsumer.minOffset(mq); long maxOffset = defaultMQPullConsumer.maxOffset(mq); if (timeValueBegin > 0) { minOffset = defaultMQPullConsumer.searchOffset(mq, timeValueBegin); } if (timeValueEnd > 0) { maxOffset = defaultMQPullConsumer.searchOffset(mq, timeValueEnd); } if (maxOffset - minOffset > countLeft) { System.out.printf("The older %d message of the %d queue will be provided\n", countLeft, mq.getQueueId()); maxOffset = minOffset + countLeft - 1; countLeft = 0; } else { countLeft = countLeft - (maxOffset - minOffset) - 1; } pullMessageByQueue(mq, minOffset, maxOffset); } } catch (Exception e) { e.printStackTrace(); } } }
ConsumeMessageCommand implements SubCommand { private void executeDefault(long timeValueBegin, long timeValueEnd) { try { Set<MessageQueue> mqs = defaultMQPullConsumer.fetchSubscribeMessageQueues(topic); long countLeft = messageCount; for (MessageQueue mq : mqs) { if (countLeft == 0) { return; } long minOffset = defaultMQPullConsumer.minOffset(mq); long maxOffset = defaultMQPullConsumer.maxOffset(mq); if (timeValueBegin > 0) { minOffset = defaultMQPullConsumer.searchOffset(mq, timeValueBegin); } if (timeValueEnd > 0) { maxOffset = defaultMQPullConsumer.searchOffset(mq, timeValueEnd); } if (maxOffset - minOffset > countLeft) { System.out.printf("The older %d message of the %d queue will be provided\n", countLeft, mq.getQueueId()); maxOffset = minOffset + countLeft - 1; countLeft = 0; } else { countLeft = countLeft - (maxOffset - minOffset) - 1; } pullMessageByQueue(mq, minOffset, maxOffset); } } catch (Exception e) { e.printStackTrace(); } }  }
ConsumeMessageCommand implements SubCommand { private void executeDefault(long timeValueBegin, long timeValueEnd) { try { Set<MessageQueue> mqs = defaultMQPullConsumer.fetchSubscribeMessageQueues(topic); long countLeft = messageCount; for (MessageQueue mq : mqs) { if (countLeft == 0) { return; } long minOffset = defaultMQPullConsumer.minOffset(mq); long maxOffset = defaultMQPullConsumer.maxOffset(mq); if (timeValueBegin > 0) { minOffset = defaultMQPullConsumer.searchOffset(mq, timeValueBegin); } if (timeValueEnd > 0) { maxOffset = defaultMQPullConsumer.searchOffset(mq, timeValueEnd); } if (maxOffset - minOffset > countLeft) { System.out.printf("The older %d message of the %d queue will be provided\n", countLeft, mq.getQueueId()); maxOffset = minOffset + countLeft - 1; countLeft = 0; } else { countLeft = countLeft - (maxOffset - minOffset) - 1; } pullMessageByQueue(mq, minOffset, maxOffset); } } catch (Exception e) { e.printStackTrace(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(final Options options); @Override void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook); }
ConsumeMessageCommand implements SubCommand { private void executeDefault(long timeValueBegin, long timeValueEnd) { try { Set<MessageQueue> mqs = defaultMQPullConsumer.fetchSubscribeMessageQueues(topic); long countLeft = messageCount; for (MessageQueue mq : mqs) { if (countLeft == 0) { return; } long minOffset = defaultMQPullConsumer.minOffset(mq); long maxOffset = defaultMQPullConsumer.maxOffset(mq); if (timeValueBegin > 0) { minOffset = defaultMQPullConsumer.searchOffset(mq, timeValueBegin); } if (timeValueEnd > 0) { maxOffset = defaultMQPullConsumer.searchOffset(mq, timeValueEnd); } if (maxOffset - minOffset > countLeft) { System.out.printf("The older %d message of the %d queue will be provided\n", countLeft, mq.getQueueId()); maxOffset = minOffset + countLeft - 1; countLeft = 0; } else { countLeft = countLeft - (maxOffset - minOffset) - 1; } pullMessageByQueue(mq, minOffset, maxOffset); } } catch (Exception e) { e.printStackTrace(); } }  @Override String commandName(); @Override String commandDesc(); @Override Options buildCommandlineOptions(final Options options); @Override void execute(final CommandLine commandLine, final Options options, RPCHook rpcHook);  }