public void handle()

in asterix-bad/src/main/java/org/apache/asterix/bad/lang/statement/ChannelSubscribeStatement.java [129:227]
80 lines of code
9 McCabe index (conditional complexity)

    public void handle(IHyracksClientConnection hcc, IStatementExecutor statementExecutor,
            IRequestParameters requestParameters, MetadataProvider metadataProvider, int resultSetId)
            throws HyracksDataException, AlgebricksException {
        DataverseName dataverse = statementExecutor.getActiveDataverseName(dataverseName);
        DataverseName brokerDataverse = statementExecutor.getActiveDataverseName(brokerDataverseName);

        MetadataTransactionContext mdTxnCtx = null;
        try {
            mdTxnCtx = MetadataManager.INSTANCE.beginTransaction();

            Channel channel = BADLangExtension.getChannel(mdTxnCtx, dataverse, channelName.getValue());
            if (channel == null) {
                throw new AsterixException("There is no channel with this name " + channelName + ".");
            }
            Broker broker = BADLangExtension.getBroker(mdTxnCtx, brokerDataverse, brokerName.getValue());
            if (broker == null) {
                throw new AsterixException("There is no broker with this name " + brokerName + ".");
            }

            String subscriptionsDatasetName = channel.getSubscriptionsDataset();

            if (argList.size() != channel.getFunction().getArity()) {
                throw new AsterixException("Channel expected " + channel.getFunction().getArity()
                        + " parameters but got " + argList.size());
            }

            Query subscriptionTuple = new Query(false);

            List<FieldBinding> fb = new ArrayList<>();
            LiteralExpr leftExpr = new LiteralExpr(new StringLiteral(BADConstants.METADATA_TYPE_NAME_DATAVERSENAME));
            Expression rightExpr = new LiteralExpr(new StringLiteral(brokerDataverse.getCanonicalForm()));
            fb.add(new FieldBinding(leftExpr, rightExpr));

            leftExpr = new LiteralExpr(new StringLiteral(BADConstants.METADATA_TYPE_FIELD_NAME_BROKERNAME));
            rightExpr = new LiteralExpr(new StringLiteral(broker.getBrokerName()));
            fb.add(new FieldBinding(leftExpr, rightExpr));

            if (subscriptionId != null) {
                leftExpr = new LiteralExpr(new StringLiteral(BADConstants.SubscriptionId));

                List<Expression> UUIDList = new ArrayList<>();
                UUIDList.add(new LiteralExpr(new StringLiteral(subscriptionId)));
                FunctionIdentifier function = BuiltinFunctions.UUID_CONSTRUCTOR;
                FunctionSignature UUIDfunc = new FunctionSignature(function);
                CallExpr UUIDCall = new CallExpr(UUIDfunc, UUIDList);

                rightExpr = UUIDCall;
                fb.add(new FieldBinding(leftExpr, rightExpr));
            }

            for (int i = 0; i < argList.size(); i++) {
                leftExpr = new LiteralExpr(new StringLiteral("param" + i));
                rightExpr = argList.get(i);
                fb.add(new FieldBinding(leftExpr, rightExpr));
            }
            RecordConstructor recordCon = new RecordConstructor(fb);
            subscriptionTuple.setBody(recordCon);
            subscriptionTuple.setVarCounter(varCounter);

            metadataProvider.setResultSetId(new ResultSetId(resultSetId));
            MetadataProvider tempMdProvider = BADUtils.replicateMetadataProvider(metadataProvider);

            final ResultDelivery resultDelivery = requestParameters.getResultProperties().getDelivery();
            final IResultSet resultSet = requestParameters.getResultSet();
            final Stats stats = requestParameters.getStats();
            if (subscriptionId == null) {
                //To create a new subscription
                VariableExpr resultVar = new VariableExpr(new VarIdentifier("$result", 0));
                VariableExpr useResultVar = new VariableExpr(new VarIdentifier("$result", 0));
                useResultVar.setIsNewVar(false);
                FieldAccessor accessor = new FieldAccessor(useResultVar, new Identifier(BADConstants.SubscriptionId));

                boolean resultsAsync =
                        resultDelivery == ResultDelivery.ASYNC || resultDelivery == ResultDelivery.DEFERRED;
                metadataProvider.setResultAsyncMode(resultsAsync);
                tempMdProvider.setResultAsyncMode(resultsAsync);
                tempMdProvider.setMaxResultReads(requestParameters.getResultProperties().getMaxReads());

                InsertStatement insert = new InsertStatement(dataverse, subscriptionsDatasetName, subscriptionTuple,
                        varCounter, resultVar, accessor);
                ((QueryTranslator) statementExecutor).handleInsertUpsertStatement(tempMdProvider, insert, hcc,
                        resultSet, resultDelivery, null, stats, false, requestParameters, null, null);
            } else {
                //To update an existing subscription
                UpsertStatement upsert = new UpsertStatement(dataverse, subscriptionsDatasetName, subscriptionTuple,
                        varCounter, null, null);
                ((QueryTranslator) statementExecutor).handleInsertUpsertStatement(tempMdProvider, upsert, hcc,
                        resultSet, resultDelivery, null, stats, false, requestParameters, null, null);
            }

            MetadataManager.INSTANCE.commitTransaction(mdTxnCtx);
        } catch (Exception e) {
            QueryTranslator.abort(e, e, mdTxnCtx);
            throw HyracksDataException.create(e);
        } finally {
            metadataProvider.getLocks().unlock();
        }

    }