@Test public void testMultipleTopicSubscribe01() throws Exception { String[] topics = {this.topicName, this.topic2, this.topic3}; String result = this.iConsumer.subscribe(topics, this.channelName, this.lastEventId, this.ext, this.defaultListener); Assert.assertFalse(result.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);  }