class 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);
        // topic pattern
        if (Subscription.isTopicPattern(topic)) {
            Subscription.validateTopicPattern(topic);
            if (StringUtils.isNumeric(offset)) {
                ParamCheckUtils.validateBlockHeight(offset, currentBlock);
            } else if (isEventId(offset)) {
                // not a topic name
                ParamCheckUtils.validateEventId("", offset, currentBlock);
            }
        } else {    // topic name
            ParamCheckUtils.validateTopicName(topic);

            // check topic exist
            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);
    }

    public  FabricBroker4Consumer(FabricDelegate fabricDelegate);

}