@Test public void testUnsubscribe01() { try { this.iConsumer.unSubscribe(null); Assert.fail(); } catch (BrokerException e) { Assert.assertEquals(e.getCode(), ErrorCode.SUBSCRIPTIONID_IS_BLANK.getCode()); } }
@Override public boolean unSubscribe(String subscriptionId) throws BrokerException { if (StringUtils.isBlank(subscriptionId)) { throw new BrokerException(ErrorCode.SUBSCRIPTIONID_IS_BLANK); } String[] tokens = subscriptionId.split(AMOPSubscription.SEPARATE); if (tokens.length == 2) { AMOPSubscription amopSubscription = this.AMOPSubscriptions.get(Long.valueOf(tokens[0])); amopSubscription.removeTopic(subscriptionId); return true; } if (!this.subscriptions.containsKey(subscriptionId)) { log.warn("not exist subscriptionId {}", subscriptionId); throw new BrokerException(ErrorCode.SUBSCRIPTIONID_NOT_EXIST); } Subscription subscription = this.subscriptions.get(subscriptionId); this.mainEventLoops.get(Long.valueOf(subscription.getGroupId())).removeSubscription(subscription); this.subscriptions.remove(subscriptionId); log.info("unSubscribe success, subscriptionId: {}", subscriptionId); return true; }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public boolean unSubscribe(String subscriptionId) throws BrokerException { if (StringUtils.isBlank(subscriptionId)) { throw new BrokerException(ErrorCode.SUBSCRIPTIONID_IS_BLANK); } String[] tokens = subscriptionId.split(AMOPSubscription.SEPARATE); if (tokens.length == 2) { AMOPSubscription amopSubscription = this.AMOPSubscriptions.get(Long.valueOf(tokens[0])); amopSubscription.removeTopic(subscriptionId); return true; } if (!this.subscriptions.containsKey(subscriptionId)) { log.warn("not exist subscriptionId {}", subscriptionId); throw new BrokerException(ErrorCode.SUBSCRIPTIONID_NOT_EXIST); } Subscription subscription = this.subscriptions.get(subscriptionId); this.mainEventLoops.get(Long.valueOf(subscription.getGroupId())).removeSubscription(subscription); this.subscriptions.remove(subscriptionId); log.info("unSubscribe success, subscriptionId: {}", subscriptionId); return true; } }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public boolean unSubscribe(String subscriptionId) throws BrokerException { if (StringUtils.isBlank(subscriptionId)) { throw new BrokerException(ErrorCode.SUBSCRIPTIONID_IS_BLANK); } String[] tokens = subscriptionId.split(AMOPSubscription.SEPARATE); if (tokens.length == 2) { AMOPSubscription amopSubscription = this.AMOPSubscriptions.get(Long.valueOf(tokens[0])); amopSubscription.removeTopic(subscriptionId); return true; } if (!this.subscriptions.containsKey(subscriptionId)) { log.warn("not exist subscriptionId {}", subscriptionId); throw new BrokerException(ErrorCode.SUBSCRIPTIONID_NOT_EXIST); } Subscription subscription = this.subscriptions.get(subscriptionId); this.mainEventLoops.get(Long.valueOf(subscription.getGroupId())).removeSubscription(subscription); this.subscriptions.remove(subscriptionId); log.info("unSubscribe success, subscriptionId: {}", subscriptionId); return true; } FiscoBcosBroker4Consumer(FiscoBcosDelegate fiscoBcosDelegate); }
FiscoBcosBroker4Consumer extends FiscoBcosTopicAdmin implements IConsumer, IBlockChain, FiscoBcosDelegate.IBlockEventListener { @Override public boolean unSubscribe(String subscriptionId) throws BrokerException { if (StringUtils.isBlank(subscriptionId)) { throw new BrokerException(ErrorCode.SUBSCRIPTIONID_IS_BLANK); } String[] tokens = subscriptionId.split(AMOPSubscription.SEPARATE); if (tokens.length == 2) { AMOPSubscription amopSubscription = this.AMOPSubscriptions.get(Long.valueOf(tokens[0])); amopSubscription.removeTopic(subscriptionId); return true; } if (!this.subscriptions.containsKey(subscriptionId)) { log.warn("not exist subscriptionId {}", subscriptionId); throw new BrokerException(ErrorCode.SUBSCRIPTIONID_NOT_EXIST); } Subscription subscription = this.subscriptions.get(subscriptionId); this.mainEventLoops.get(Long.valueOf(subscription.getGroupId())).removeSubscription(subscription); this.subscriptions.remove(subscriptionId); log.info("unSubscribe success, subscriptionId: {}", subscriptionId); return true; } 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 boolean unSubscribe(String subscriptionId) throws BrokerException { if (StringUtils.isBlank(subscriptionId)) { throw new BrokerException(ErrorCode.SUBSCRIPTIONID_IS_BLANK); } String[] tokens = subscriptionId.split(AMOPSubscription.SEPARATE); if (tokens.length == 2) { AMOPSubscription amopSubscription = this.AMOPSubscriptions.get(Long.valueOf(tokens[0])); amopSubscription.removeTopic(subscriptionId); return true; } if (!this.subscriptions.containsKey(subscriptionId)) { log.warn("not exist subscriptionId {}", subscriptionId); throw new BrokerException(ErrorCode.SUBSCRIPTIONID_NOT_EXIST); } Subscription subscription = this.subscriptions.get(subscriptionId); this.mainEventLoops.get(Long.valueOf(subscription.getGroupId())).removeSubscription(subscription); this.subscriptions.remove(subscriptionId); log.info("unSubscribe success, subscriptionId: {}", subscriptionId); return true; } 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);  }