当前位置：源码 > Java源码 >

Java源码示例：org.apache.rocketmq.common.message.MessageClientExt

示例1

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListener().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListener().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例2

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListener().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListener().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例3

private ArrayList<MessageExt> createMessageList(int size) {
    ArrayList<MessageExt> list = new ArrayList<>();
    for (int i = 0; i < size; i++) {
        MessageClientExt messageClientExt = new MessageClientExt();
        messageClientExt.setTopic(topic);
        messageClientExt.setQueueId(0);
        messageClientExt.setMsgId("123");
        messageClientExt.setBody(new byte[] {'a'});
        messageClientExt.setOffsetMsgId("234");
        messageClientExt.setBornHost(new InetSocketAddress(8080));
        messageClientExt.setStoreHost(new InetSocketAddress(8080));
        messageClientExt.putUserProperty(DeFiBusConstant.PROPERTY_MESSAGE_TTL, "3000");
        list.add(messageClientExt);
    }
    return list;
}

示例4

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListener().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListener().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例5

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListner().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListner().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListner().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMonment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例6

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListener().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListener().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例7

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListner().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListner().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListner().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例8

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListner().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListner().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListner().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMonment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例9

@Test
public void testQueryMsg() {
    int msgSize = 20;
    producer.send(msgSize);
    Assert.assertEquals("Not all are sent", msgSize, producer.getAllUndupMsgBody().size());
    consumer.getListener().waitForMessageConsume(producer.getAllMsgBody(), consumeTime);
    Assert.assertEquals("Not all are consumed", 0, VerifyUtils.verify(producer.getAllMsgBody(),
        consumer.getListener().getAllMsgBody()));

    MessageExt recvMsg = (MessageExt) consumer.getListener().getFirstMsg();
    MessageExt queryMsg = null;
    try {
        TestUtils.waitForMoment(3000);
        queryMsg = producer.getProducer().viewMessage(((MessageClientExt) recvMsg).getOffsetMsgId());
    } catch (Exception e) {
    }

    assertThat(queryMsg).isNotNull();
    assertThat(new String(queryMsg.getBody())).isEqualTo(new String(recvMsg.getBody()));
}

示例10

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
    final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例11

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
    final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例12

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
    final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例13

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId, final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例14

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
    final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例15

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId, final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例16

@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
    final String topic,
    final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgId, timeoutMillis * 3);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(RemotingUtil.socketAddress2String(msg.getStoreHost()),
            consumerGroup, clientId, msgClient.getOffsetMsgId(), timeoutMillis * 3);
    }
}

示例17

@Override
public boolean resumeCheckHalfMessage(final String topic, final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
    MessageExt msg = this.viewMessage(topic, msgId);
    if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
        return this.mqClientInstance.getMQClientAPIImpl().resumeCheckHalfMessage(RemotingUtil.socketAddress2String(msg.getStoreHost()), msgId, timeoutMillis);
    } else {
        MessageClientExt msgClient = (MessageClientExt) msg;
        return this.mqClientInstance.getMQClientAPIImpl().resumeCheckHalfMessage(RemotingUtil.socketAddress2String(msg.getStoreHost()), msgClient.getOffsetMsgId(), timeoutMillis);
    }
}

示例18

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例19

public static void printMsg(final MessageExt msg) throws IOException {
    if (msg == null) {
        System.out.printf("%nMessage not found!");
        return;
    }

    String bodyTmpFilePath = createBodyFile(msg);
    String msgId = msg.getMsgId();
    if (msg instanceof MessageClientExt) {
        msgId = ((MessageClientExt) msg).getOffsetMsgId();
    }

    System.out.printf("%-20s %s%n",
        "OffsetID:",
        msgId
    );

    System.out.printf("%-20s %s%n",
        "Topic:",
        msg.getTopic()
    );

    System.out.printf("%-20s %s%n",
        "Tags:",
        "[" + msg.getTags() + "]"
    );

    System.out.printf("%-20s %s%n",
        "Keys:",
        "[" + msg.getKeys() + "]"
    );

    System.out.printf("%-20s %d%n",
        "Queue ID:",
        msg.getQueueId()
    );

    System.out.printf("%-20s %d%n",
        "Queue Offset:",
        msg.getQueueOffset()
    );

    System.out.printf("%-20s %d%n",
        "CommitLog Offset:",
        msg.getCommitLogOffset()
    );

    System.out.printf("%-20s %d%n",
        "Reconsume Times:",
        msg.getReconsumeTimes()
    );

    System.out.printf("%-20s %s%n",
        "Born Timestamp:",
        UtilAll.timeMillisToHumanString2(msg.getBornTimestamp())
    );

    System.out.printf("%-20s %s%n",
        "Store Timestamp:",
        UtilAll.timeMillisToHumanString2(msg.getStoreTimestamp())
    );

    System.out.printf("%-20s %s%n",
        "Born Host:",
        RemotingHelper.parseSocketAddressAddr(msg.getBornHost())
    );

    System.out.printf("%-20s %s%n",
        "Store Host:",
        RemotingHelper.parseSocketAddressAddr(msg.getStoreHost())
    );

    System.out.printf("%-20s %d%n",
        "System Flag:",
        msg.getSysFlag()
    );

    System.out.printf("%-20s %s%n",
        "Properties:",
        msg.getProperties() != null ? msg.getProperties().toString() : ""
    );

    System.out.printf("%-20s %s%n",
        "Message Body Path:",
        bodyTmpFilePath
    );
}

示例20

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例21

@Before
public void init() throws Exception {
    DeFiBusClientConfig clientConfig = new DeFiBusClientConfig();
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    clientConfig.setConsumerGroup(consumerGroup);
    pushConsumer = new DeFiBusPushConsumer(clientConfig);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.getDefaultMQPushConsumer().setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new DeFiBusMessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus handleMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            messageExts[0] = msgs.get(0);
            countDownLatch.countDown();
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumer().getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumer().getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic);
    pushConsumer.start();

    defiBusClientFactory = spy(pushConsumer.getDeFiBusClientInstance());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, defiBusClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(defiBusClientFactory, deFiBusClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(defiBusClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumer().getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(defiBusClientFactory);
    defiBusClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(defiBusClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                messageClientExt.putUserProperty(DeFiBusConstant.PROPERTY_MESSAGE_TTL, "3000");
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10912", false)).when(defiBusClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumer().getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
}

示例22

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例23

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);


    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
        @Override public Object answer(InvocationOnMock mock) throws Throwable {
            PullMessageRequestHeader requestHeader = mock.getArgument(1);
            MessageClientExt messageClientExt = new MessageClientExt();
            messageClientExt.setTopic(topic);
            messageClientExt.setQueueId(0);
            messageClientExt.setMsgId("123");
            messageClientExt.setBody(new byte[] {'a'});
            messageClientExt.setOffsetMsgId("234");
            messageClientExt.setBornHost(new InetSocketAddress(8080));
            messageClientExt.setStoreHost(new InetSocketAddress(8080));
            PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
            ((PullCallback)mock.getArgument(4)).onSuccess(pullResult);
            return pullResult;
        }
    });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例24

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例25

public static void printMsg(final MessageExt msg) throws IOException {
    if (msg == null) {
        System.out.printf("%nMessage not found!");
        return;
    }

    String bodyTmpFilePath = createBodyFile(msg);
    String msgId = msg.getMsgId();
    if (msg instanceof MessageClientExt) {
        msgId = ((MessageClientExt) msg).getOffsetMsgId();
    }

    System.out.printf("%-20s %s%n",
        "OffsetID:",
        msgId
    );

    System.out.printf("%-20s %s%n",
        "Topic:",
        msg.getTopic()
    );

    System.out.printf("%-20s %s%n",
        "Tags:",
        "[" + msg.getTags() + "]"
    );

    System.out.printf("%-20s %s%n",
        "Keys:",
        "[" + msg.getKeys() + "]"
    );

    System.out.printf("%-20s %d%n",
        "Queue ID:",
        msg.getQueueId()
    );

    System.out.printf("%-20s %d%n",
        "Queue Offset:",
        msg.getQueueOffset()
    );

    System.out.printf("%-20s %d%n",
        "CommitLog Offset:",
        msg.getCommitLogOffset()
    );

    System.out.printf("%-20s %d%n",
        "Reconsume Times:",
        msg.getReconsumeTimes()
    );

    System.out.printf("%-20s %s%n",
        "Born Timestamp:",
        UtilAll.timeMillisToHumanString2(msg.getBornTimestamp())
    );

    System.out.printf("%-20s %s%n",
        "Store Timestamp:",
        UtilAll.timeMillisToHumanString2(msg.getStoreTimestamp())
    );

    System.out.printf("%-20s %s%n",
        "Born Host:",
        RemotingHelper.parseSocketAddressAddr(msg.getBornHost())
    );

    System.out.printf("%-20s %s%n",
        "Store Host:",
        RemotingHelper.parseSocketAddressAddr(msg.getStoreHost())
    );

    System.out.printf("%-20s %d%n",
        "System Flag:",
        msg.getSysFlag()
    );

    System.out.printf("%-20s %s%n",
        "Properties:",
        msg.getProperties() != null ? msg.getProperties().toString() : ""
    );

    System.out.printf("%-20s %s%n",
        "Message Body Path:",
        bodyTmpFilePath
    );
}

示例26

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);


    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
        @Override public Object answer(InvocationOnMock mock) throws Throwable {
            PullMessageRequestHeader requestHeader = mock.getArgument(1);
            MessageClientExt messageClientExt = new MessageClientExt();
            messageClientExt.setTopic(topic);
            messageClientExt.setQueueId(0);
            messageClientExt.setMsgId("123");
            messageClientExt.setBody(new byte[] {'a'});
            messageClientExt.setOffsetMsgId("234");
            messageClientExt.setBornHost(new InetSocketAddress(8080));
            messageClientExt.setStoreHost(new InetSocketAddress(8080));
            PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
            ((PullCallback)mock.getArgument(4)).onSuccess(pullResult);
            return pullResult;
        }
    });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例27

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);


    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
        @Override public Object answer(InvocationOnMock mock) throws Throwable {
            PullMessageRequestHeader requestHeader = mock.getArgument(1);
            MessageClientExt messageClientExt = new MessageClientExt();
            messageClientExt.setTopic(topic);
            messageClientExt.setQueueId(0);
            messageClientExt.setMsgId("123");
            messageClientExt.setBody(new byte[] {'a'});
            messageClientExt.setOffsetMsgId("234");
            messageClientExt.setBornHost(new InetSocketAddress(8080));
            messageClientExt.setStoreHost(new InetSocketAddress(8080));
            PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
            ((PullCallback)mock.getArgument(4)).onSuccess(pullResult);
            return pullResult;
        }
    });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例28

private void initDefaultLitePullConsumer(DefaultLitePullConsumer litePullConsumer) throws Exception {

        Field field = DefaultLitePullConsumer.class.getDeclaredField("defaultLitePullConsumerImpl");
        field.setAccessible(true);
        litePullConsumerImpl = (DefaultLitePullConsumerImpl) field.get(litePullConsumer);
        field = DefaultLitePullConsumerImpl.class.getDeclaredField("mQClientFactory");
        field.setAccessible(true);
        field.set(litePullConsumerImpl, mQClientFactory);

        PullAPIWrapper pullAPIWrapper = litePullConsumerImpl.getPullAPIWrapper();
        field = PullAPIWrapper.class.getDeclaredField("mQClientFactory");
        field.setAccessible(true);
        field.set(pullAPIWrapper, mQClientFactory);

        field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
        field.setAccessible(true);
        field.set(mQClientFactory, mQClientAPIImpl);

        field = MQClientInstance.class.getDeclaredField("mQAdminImpl");
        field.setAccessible(true);
        field.set(mQClientFactory, mQAdminImpl);

        field = DefaultLitePullConsumerImpl.class.getDeclaredField("rebalanceImpl");
        field.setAccessible(true);
        rebalanceImpl = (RebalanceImpl) field.get(litePullConsumerImpl);
        field = RebalanceImpl.class.getDeclaredField("mQClientFactory");
        field.setAccessible(true);
        field.set(rebalanceImpl, mQClientFactory);

        offsetStore = spy(litePullConsumerImpl.getOffsetStore());
        field = DefaultLitePullConsumerImpl.class.getDeclaredField("offsetStore");
        field.setAccessible(true);
        field.set(litePullConsumerImpl, offsetStore);

        when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
            anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
            .thenAnswer(new Answer<Object>() {
                @Override
                public Object answer(InvocationOnMock mock) throws Throwable {
                    PullMessageRequestHeader requestHeader = mock.getArgument(1);
                    MessageClientExt messageClientExt = new MessageClientExt();
                    messageClientExt.setTopic(topic);
                    messageClientExt.setQueueId(0);
                    messageClientExt.setMsgId("123");
                    messageClientExt.setBody(new byte[] {'a'});
                    messageClientExt.setOffsetMsgId("234");
                    messageClientExt.setBornHost(new InetSocketAddress(8080));
                    messageClientExt.setStoreHost(new InetSocketAddress(8080));
                    PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                    return pullResult;
                }
            });

        when(mQClientFactory.findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean())).thenReturn(new FindBrokerResult("127.0.0.1:10911", false));

        doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());

        doReturn(123L).when(offsetStore).readOffset(any(MessageQueue.class), any(ReadOffsetType.class));
    }

示例29

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    PowerMockito.suppress(PowerMockito.method(DefaultMQPushConsumerImpl.class, "updateTopicSubscribeInfoWhenSubscriptionChanged"));
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);

    pushConsumer.subscribe(topic, "*");
    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    doReturn(Collections.singletonList(mQClientFactory.getClientId())).when(mQClientFactory).findConsumerIdList(anyString(), anyString());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
    doReturn(123L).when(rebalancePushImpl).computePullFromWhere(any(MessageQueue.class));
}

示例30

@Before
public void init() throws Exception {
    consumerGroup = "FooBarGroup" + System.currentTimeMillis();
    pushConsumer = new DefaultMQPushConsumer(consumerGroup, true, "");
    consumerGroupNormal = "FooBarGroup" + System.currentTimeMillis();
    normalPushConsumer = new DefaultMQPushConsumer(consumerGroupNormal, false, "");
    customTraceTopicpushConsumer = new DefaultMQPushConsumer(consumerGroup, true, customerTraceTopic);
    pushConsumer.setNamesrvAddr("127.0.0.1:9876");
    pushConsumer.setPullInterval(60 * 1000);

    asyncTraceDispatcher = (AsyncTraceDispatcher) pushConsumer.getTraceDispatcher();
    traceProducer = asyncTraceDispatcher.getTraceProducer();

    pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
            ConsumeConcurrentlyContext context) {
            return null;
        }
    });

    PowerMockito.suppress(PowerMockito.method(DefaultMQPushConsumerImpl.class, "updateTopicSubscribeInfoWhenSubscriptionChanged"));
    DefaultMQPushConsumerImpl pushConsumerImpl = pushConsumer.getDefaultMQPushConsumerImpl();
    rebalancePushImpl = spy(new RebalancePushImpl(pushConsumer.getDefaultMQPushConsumerImpl()));
    Field field = DefaultMQPushConsumerImpl.class.getDeclaredField("rebalanceImpl");
    field.setAccessible(true);
    field.set(pushConsumerImpl, rebalancePushImpl);
    pushConsumer.subscribe(topic, "*");

    pushConsumer.start();

    mQClientFactory = spy(pushConsumerImpl.getmQClientFactory());
    mQClientTraceFactory = spy(pushConsumerImpl.getmQClientFactory());

    field = DefaultMQPushConsumerImpl.class.getDeclaredField("mQClientFactory");
    field.setAccessible(true);
    field.set(pushConsumerImpl, mQClientFactory);

    field = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    field.setAccessible(true);
    field.set(mQClientFactory, mQClientAPIImpl);

    Field fieldTrace = DefaultMQProducerImpl.class.getDeclaredField("mQClientFactory");
    fieldTrace.setAccessible(true);
    fieldTrace.set(traceProducer.getDefaultMQProducerImpl(), mQClientTraceFactory);

    fieldTrace = MQClientInstance.class.getDeclaredField("mQClientAPIImpl");
    fieldTrace.setAccessible(true);
    fieldTrace.set(mQClientTraceFactory, mQClientTraceAPIImpl);

    pullAPIWrapper = spy(new PullAPIWrapper(mQClientFactory, consumerGroup, false));
    field = DefaultMQPushConsumerImpl.class.getDeclaredField("pullAPIWrapper");
    field.setAccessible(true);
    field.set(pushConsumerImpl, pullAPIWrapper);

    pushConsumer.getDefaultMQPushConsumerImpl().getRebalanceImpl().setmQClientFactory(mQClientFactory);
    mQClientFactory.registerConsumer(consumerGroup, pushConsumerImpl);

    when(mQClientFactory.getMQClientAPIImpl().pullMessage(anyString(), any(PullMessageRequestHeader.class),
        anyLong(), any(CommunicationMode.class), nullable(PullCallback.class)))
        .thenAnswer(new Answer<Object>() {
            @Override
            public Object answer(InvocationOnMock mock) throws Throwable {
                PullMessageRequestHeader requestHeader = mock.getArgument(1);
                MessageClientExt messageClientExt = new MessageClientExt();
                messageClientExt.setTopic(topic);
                messageClientExt.setQueueId(0);
                messageClientExt.setMsgId("123");
                messageClientExt.setBody(new byte[] {'a'});
                messageClientExt.setOffsetMsgId("234");
                messageClientExt.setBornHost(new InetSocketAddress(8080));
                messageClientExt.setStoreHost(new InetSocketAddress(8080));
                PullResult pullResult = createPullResult(requestHeader, PullStatus.FOUND, Collections.<MessageExt>singletonList(messageClientExt));
                ((PullCallback) mock.getArgument(4)).onSuccess(pullResult);
                return pullResult;
            }
        });

    doReturn(new FindBrokerResult("127.0.0.1:10911", false)).when(mQClientFactory).findBrokerAddressInSubscribe(anyString(), anyLong(), anyBoolean());
    Set<MessageQueue> messageQueueSet = new HashSet<MessageQueue>();
    messageQueueSet.add(createPullRequest().getMessageQueue());
    pushConsumer.getDefaultMQPushConsumerImpl().updateTopicSubscribeInfo(topic, messageQueueSet);
}

Java源码示例：org.apache.rocketmq.common.message.MessageClientExt

微信关注