@Test public void testSingleTopicSubscribeBoolean01() throws Exception { MyConsumerListener listener = new MyConsumerListener(); String result = this.iConsumer.subscribe(this.topicName, this.channelName, WeEvent.OFFSET_FIRST, this.ext, listener); Assert.assertFalse(result.isEmpty()); SendResult sendResult = this.iProducer.publish(new WeEvent(this.topicName, "hello world.".getBytes()), this.channelName).get(transactionTimeout, TimeUnit.MILLISECONDS); Assert.assertEquals(SendResult.SendResultStatus.SUCCESS, sendResult.getStatus()); Thread.sleep(wait3s * 10); Assert.assertFalse(listener.notifiedEvents.isEmpty()); }
@Override public String subscribe(String topic, String channelName, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { this.validateChannelName(channelName); ParamCheckUtils.validateOffset(offset); Long currentBlock = fabricDelegate.getBlockHeight(channelName); if (Subscription.isTopicPattern(topic)) { Subscription.validateTopicPattern(topic); if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId("", offset, currentBlock); } } else { ParamCheckUtils.validateTopicName(topic); if (!this.exist(topic, channelName)) { throw new BrokerException(ErrorCode.TOPIC_NOT_EXIST); } if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId(topic, offset, currentBlock); } } log.info("subscribe channelName: {} topic: {} offset: {} ext: {}", channelName, topic, offset, ext); return subscribeTopic(topic, channelName, offset, ext, listener); }
FabricBroker4Consumer extends FabricTopicAdmin implements IConsumer, IBlockChain { @Override public String subscribe(String topic, String channelName, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { this.validateChannelName(channelName); ParamCheckUtils.validateOffset(offset); Long currentBlock = fabricDelegate.getBlockHeight(channelName); if (Subscription.isTopicPattern(topic)) { Subscription.validateTopicPattern(topic); if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId("", offset, currentBlock); } } else { ParamCheckUtils.validateTopicName(topic); if (!this.exist(topic, channelName)) { throw new BrokerException(ErrorCode.TOPIC_NOT_EXIST); } if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId(topic, offset, currentBlock); } } log.info("subscribe channelName: {} topic: {} offset: {} ext: {}", channelName, topic, offset, ext); return subscribeTopic(topic, channelName, offset, ext, listener); } }
FabricBroker4Consumer extends FabricTopicAdmin implements IConsumer, IBlockChain { @Override public String subscribe(String topic, String channelName, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { this.validateChannelName(channelName); ParamCheckUtils.validateOffset(offset); Long currentBlock = fabricDelegate.getBlockHeight(channelName); if (Subscription.isTopicPattern(topic)) { Subscription.validateTopicPattern(topic); if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId("", offset, currentBlock); } } else { ParamCheckUtils.validateTopicName(topic); if (!this.exist(topic, channelName)) { throw new BrokerException(ErrorCode.TOPIC_NOT_EXIST); } if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId(topic, offset, currentBlock); } } log.info("subscribe channelName: {} topic: {} offset: {} ext: {}", channelName, topic, offset, ext); return subscribeTopic(topic, channelName, offset, ext, listener); } FabricBroker4Consumer(FabricDelegate fabricDelegate); }
FabricBroker4Consumer extends FabricTopicAdmin implements IConsumer, IBlockChain { @Override public String subscribe(String topic, String channelName, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { this.validateChannelName(channelName); ParamCheckUtils.validateOffset(offset); Long currentBlock = fabricDelegate.getBlockHeight(channelName); if (Subscription.isTopicPattern(topic)) { Subscription.validateTopicPattern(topic); if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId("", offset, currentBlock); } } else { ParamCheckUtils.validateTopicName(topic); if (!this.exist(topic, channelName)) { throw new BrokerException(ErrorCode.TOPIC_NOT_EXIST); } if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId(topic, offset, currentBlock); } } log.info("subscribe channelName: {} topic: {} offset: {} ext: {}", channelName, topic, offset, ext); return subscribeTopic(topic, channelName, offset, ext, listener); } FabricBroker4Consumer(FabricDelegate fabricDelegate); @Override String subscribe(String topic, String channelName, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override String subscribe(String[] topics, String channelName, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override boolean unSubscribe(String subscriptionId); @Override Map<String, SubscriptionInfo> listSubscription(String channnelName); @Override boolean isStarted(); @Override synchronized boolean startConsumer(); @Override synchronized boolean shutdownConsumer(); @Override int getIdleTime(); @Override Long getBlockHeight(String channelName); @Override boolean hasBlockEventNotify(); @Override List<WeEvent> loop(Long blockNum, String channelName); }
FabricBroker4Consumer extends FabricTopicAdmin implements IConsumer, IBlockChain { @Override public String subscribe(String topic, String channelName, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { this.validateChannelName(channelName); ParamCheckUtils.validateOffset(offset); Long currentBlock = fabricDelegate.getBlockHeight(channelName); if (Subscription.isTopicPattern(topic)) { Subscription.validateTopicPattern(topic); if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId("", offset, currentBlock); } } else { ParamCheckUtils.validateTopicName(topic); if (!this.exist(topic, channelName)) { throw new BrokerException(ErrorCode.TOPIC_NOT_EXIST); } if (StringUtils.isNumeric(offset)) { ParamCheckUtils.validateBlockHeight(offset, currentBlock); } else if (isEventId(offset)) { ParamCheckUtils.validateEventId(topic, offset, currentBlock); } } log.info("subscribe channelName: {} topic: {} offset: {} ext: {}", channelName, topic, offset, ext); return subscribeTopic(topic, channelName, offset, ext, listener); } FabricBroker4Consumer(FabricDelegate fabricDelegate); @Override String subscribe(String topic, String channelName, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override String subscribe(String[] topics, String channelName, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override boolean unSubscribe(String subscriptionId); @Override Map<String, SubscriptionInfo> listSubscription(String channnelName); @Override boolean isStarted(); @Override synchronized boolean startConsumer(); @Override synchronized boolean shutdownConsumer(); @Override int getIdleTime(); @Override Long getBlockHeight(String channelName); @Override boolean hasBlockEventNotify(); @Override List<WeEvent> loop(Long blockNum, String channelName);  }