@Test public void testSingleTopicSubscribeBoolean01() throws Exception { MyConsumerListener listener = new MyConsumerListener(); String result = this.iConsumer.subscribe(this.topicName, this.groupId, WeEvent.OFFSET_FIRST, this.ext, listener); Assert.assertFalse(result.isEmpty()); SendResult sendResult = this.iProducer.publish(new WeEvent(this.topicName, "hello world.".getBytes()), this.groupId).get(transactionTimeout, TimeUnit.SECONDS); Assert.assertEquals(SendResult.SendResultStatus.SUCCESS, sendResult.getStatus()); Thread.sleep(wait3s * 10); Assert.assertFalse(listener.notifiedEvents.isEmpty()); }
@Override public String subscribe(String topic, String groupIdStr, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { String groupId = selectGroupId(groupIdStr); this.validateGroupId(groupId); ParamCheckUtils.validateOffset(offset); Long currentBlock = fiscoBcosDelegate.getBlockHeight(Long.parseLong(groupId)); 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, groupId)) { 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 groupId: {} topic: {} offset: {} ext: {}", groupId, topic, offset, ext); return subscribeTopic(topic, Long.valueOf(groupId), offset, ext, listener); }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public String subscribe(String topic, String groupIdStr, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { String groupId = selectGroupId(groupIdStr); this.validateGroupId(groupId); ParamCheckUtils.validateOffset(offset); Long currentBlock = fiscoBcosDelegate.getBlockHeight(Long.parseLong(groupId)); 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, groupId)) { 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 groupId: {} topic: {} offset: {} ext: {}", groupId, topic, offset, ext); return subscribeTopic(topic, Long.valueOf(groupId), offset, ext, listener); } }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public String subscribe(String topic, String groupIdStr, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { String groupId = selectGroupId(groupIdStr); this.validateGroupId(groupId); ParamCheckUtils.validateOffset(offset); Long currentBlock = fiscoBcosDelegate.getBlockHeight(Long.parseLong(groupId)); 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, groupId)) { 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 groupId: {} topic: {} offset: {} ext: {}", groupId, topic, offset, ext); return subscribeTopic(topic, Long.valueOf(groupId), offset, ext, listener); } FiscoBcosBroker4Consumer(FiscoBcosDelegate fiscoBcosDelegate); }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public String subscribe(String topic, String groupIdStr, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { String groupId = selectGroupId(groupIdStr); this.validateGroupId(groupId); ParamCheckUtils.validateOffset(offset); Long currentBlock = fiscoBcosDelegate.getBlockHeight(Long.parseLong(groupId)); 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, groupId)) { 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 groupId: {} topic: {} offset: {} ext: {}", groupId, topic, offset, ext); return subscribeTopic(topic, Long.valueOf(groupId), offset, ext, listener); } FiscoBcosBroker4Consumer(FiscoBcosDelegate fiscoBcosDelegate); @Override String subscribe(String topic, String groupIdStr, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override String subscribe(String[] topics, String groupIdStr, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override boolean unSubscribe(String subscriptionId); @Override boolean isStarted(); @Override synchronized boolean startConsumer(); @Override synchronized boolean shutdownConsumer(); @Override synchronized Map<String, SubscriptionInfo> listSubscription(String groupIdStr); @Override int getIdleTime(); @Override boolean hasBlockEventNotify(); @Override List<WeEvent> loop(Long blockNum, String groupId); @Override void onEvent(Long groupId, Long blockHeight); }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public String subscribe(String topic, String groupIdStr, String offset, @NonNull Map<SubscribeExt, String> ext, @NonNull ConsumerListener listener) throws BrokerException { String groupId = selectGroupId(groupIdStr); this.validateGroupId(groupId); ParamCheckUtils.validateOffset(offset); Long currentBlock = fiscoBcosDelegate.getBlockHeight(Long.parseLong(groupId)); 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, groupId)) { 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 groupId: {} topic: {} offset: {} ext: {}", groupId, topic, offset, ext); return subscribeTopic(topic, Long.valueOf(groupId), offset, ext, listener); } FiscoBcosBroker4Consumer(FiscoBcosDelegate fiscoBcosDelegate); @Override String subscribe(String topic, String groupIdStr, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override String subscribe(String[] topics, String groupIdStr, String offset,
                            @NonNull Map<SubscribeExt, String> ext,
                            @NonNull ConsumerListener listener); @Override boolean unSubscribe(String subscriptionId); @Override boolean isStarted(); @Override synchronized boolean startConsumer(); @Override synchronized boolean shutdownConsumer(); @Override synchronized Map<String, SubscriptionInfo> listSubscription(String groupIdStr); @Override int getIdleTime(); @Override boolean hasBlockEventNotify(); @Override List<WeEvent> loop(Long blockNum, String groupId); @Override void onEvent(Long groupId, Long blockHeight);  }