public void start()

in ons-core/ons-api/src/main/java/org/apache/rocketmq/ons/api/bean/ConsumerBean.java [44:91]
38 lines of code
12 McCabe index (conditional complexity)

    public void start() {
        if (null == this.properties) {
            throw new ONSClientException("properties not set");
        }

        if (null == this.subscriptionTable) {
            throw new ONSClientException("subscriptionTable not set");
        }

        this.consumer = ONSFactory.createConsumer(this.properties);

        Iterator<Entry<Subscription, MessageListener>> it = this.subscriptionTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Subscription, MessageListener> next = it.next();
            if ("com.aliyun.openservices.ons.api.impl.notify.ConsumerImpl".equals(this.consumer.getClass().getCanonicalName())
                && (next.getKey() instanceof SubscriptionExt)) {
                SubscriptionExt subscription = (SubscriptionExt) next.getKey();
                for (Method method : this.consumer.getClass().getMethods()) {
                    if ("subscribeNotify".equals(method.getName())) {
                        try {
                            method.invoke(consumer, subscription.getTopic(), subscription.getExpression(),
                                    subscription.isPersistence(), next.getValue());
                        } catch (Exception e) {
                            throw new ONSClientException("subscribeNotify invoke exception", e);
                        }
                        break;
                    }
                }

            } else {
                Subscription subscription = next.getKey();
                if (subscription.getType() == null || ExpressionType.TAG.name().equals(subscription.getType())) {

                    this.subscribe(subscription.getTopic(), subscription.getExpression(), next.getValue());

                } else if (ExpressionType.SQL92.name().equals(subscription.getType())) {

                    this.subscribe(subscription.getTopic(), MessageSelector.bySql(subscription.getExpression()), next.getValue());
                } else {

                    throw new ONSClientException(String.format("Expression type %s is unknown!", subscription.getType()));
                }
            }

        }

        this.consumer.start();
    }