ActiveMq subscribe frame.

@param string $destination
@param string $subscriptionId
@param string $ack
@param string $selector
@param boolean|false $durable durable subscription
@return Frame
972296
public function getSubscribeFrame(
        $destination,
        $subscriptionId = null,
        $ack = 'auto',
        $selector = null,
        $durable = false
    ) {
        $frame = parent::getSubscribeFrame($destination, $subscriptionId, $ack, $selector);
        $frame['activemq.prefetchSize'] = $this->prefetchSize;
        if ($durable) {
            $frame['activemq.subscriptionName'] = $this->getClientId();
        }
        return $frame;
    }
994350
protected function receiveInternal($subscriber_id = null, $limit = -1, $mode = self::GET_RESERVE)
    {
        $primaryKey = models\DbMessage::primaryKey();
        $trx = models\DbMessage::getDb()->transaction !== null ? null : models\DbMessage::getDb()->beginTransaction();
        $messages = models\DbMessage::find()
            ->withQueue($this->id)
            ->withSubscriber($subscriber_id)
            ->available($this->timeout)
            ->limit($limit)
            ->indexBy($primaryKey[0])
            ->all();
        if (!empty($messages)) {
            $now = new \DateTime('now', new \DateTimezone('UTC'));
            if ($mode === self::GET_DELETE) {
                $attributes = ['status' => Message::DELETED, 'deleted_on' => $now->format('Y-m-d H:i:s')];
            } elseif ($mode === self::GET_RESERVE) {
                $attributes = ['status' => Message::RESERVED, 'reserved_on' => $now->format('Y-m-d H:i:s')];
            }
            if (isset($attributes)) {
                models\DbMessage::updateAll($attributes, ['in', models\DbMessage::primaryKey(), array_keys($messages)]);
            }
        }
        if ($trx !== null) {
            $trx->commit();
        }

        return models\DbMessage::createMessages($messages);
    }
1154422
@Override
    public MPSubscription getSubscription(String subscriptionName)
                    throws SIDurableSubscriptionNotFoundException
    {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "getSubscription",
                        subscriptionName);

        HashMap durableSubs = _destinationManager.getDurableSubscriptionsTable();

        ConsumerDispatcher cd = null;
        synchronized (durableSubs)
        {
            //Look up the consumer dispatcher for this subId in the system durable subs list
            cd =
                            (ConsumerDispatcher) durableSubs.get(subscriptionName);

            // Check that the durable subscription exists
            if (cd == null)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "getSubscription", "Durable sub not found");

                throw new SIDurableSubscriptionNotFoundException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_DOESNT_EXIST_ERROR_CWSIP0072",
                                                        new Object[] { subscriptionName,
                                                                      _messageProcessor.getMessagingEngineName() },
                                                        null));
            }
        }

        MPSubscription mpSubscription = cd.getMPSubscription();
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "getSubscription", mpSubscription);
        return mpSubscription;
    }
1012196
@Override
	protected final void sendToDestination(Destination destination, boolean destinationOverride, Message srcMessage, int deliveryMode, int priority, long timeToLive) throws JMSException
    {
        // Check that the destination was specified
        if (destination == null)
            throw new InvalidDestinationException("Destination not specified");  // [JMS SPEC]

        // Create an internal copy if necessary
        AbstractMessage message = MessageTools.makeInternalCopy(srcMessage);
        
        externalAccessLock.readLock().lock();
        try
		{
    		checkNotClosed();
    		
	        // Dispatch to session
            ((LocalSession)session).dispatch(message);
		}
        finally
        {
        	externalAccessLock.readLock().unlock();
        }
    }
1181995
public function exchangeBind($destination, $source, $routingKey = '', $nowait = false, $arguments = [])
    {
        return $this->getClient()->exchangeBind($this->getChannelId(), $destination, $source, $routingKey, $nowait, $arguments);
    }
1116630
public function createConsumer(Destination $destination): Consumer
    {
        InvalidDestinationException::assertDestinationInstanceOf($destination, PheanstalkDestination::class);

        return new PheanstalkConsumer($destination, $this->pheanstalk);
    }
485597
function subscribeNode (client, subscriptions, channel, callback, done_callback) {
    if (subscriptions[channel]===undefined) {
        subscriptions[channel] = [callback];
        client.subscribe(channel, {qos: 0}, function() {
            // If there is a done_callback defined, execute it
            if (done_callback!==undefined) done_callback();
        });
    } else {
        subscriptions[channel].push(callback);
    }
}
416666
public function subscribe($destination, $subscriptionId = null, $ack = 'auto', $selector = null, array $header = [])
    {
        return $this->client->sendFrame(
            $this->getProtocol()->getSubscribeFrame($destination, $subscriptionId, $ack, $selector)->addHeaders($header)
        );
    }
341413
public function createSubscriber($mode, $dsn = null, $subscribe = '', $options = array())
    {
        $options[ZMQ::SOCKOPT_SUBSCRIBE] = $subscribe;

        return $this->createSocket(ZMQ::SOCKET_SUB, $mode, $dsn, $options);
    }
1363413
public function getSubscribeFrame($destination, $subscriptionId = null, $ack = 'auto', $selector = null)
    {
        // validate ACK types per spec
        // https://stomp.github.io/stomp-specification-1.0.html#frame-ACK
        // https://stomp.github.io/stomp-specification-1.1.html#ACK
        // https://stomp.github.io/stomp-specification-1.2.html#ACK
        if ($this->hasVersion(Version::VERSION_1_1)) {
            $validAcks = ['auto', 'client', 'client-individual'];
        } else {
            $validAcks = ['auto', 'client'];
        }
        if (!in_array($ack, $validAcks)) {
            throw new StompException(
                sprintf(
                    '"%s" is not a valid ack value for STOMP %s. A valid value is one of %s',
                    $ack,
                    $this->version,
                    implode(',', $validAcks)
                )
            );
        }
        
        $frame = $this->createFrame('SUBSCRIBE');

        $frame['destination'] = $destination;
        $frame['ack'] = $ack;
        $frame['id'] = $subscriptionId;
        $frame['selector'] = $selector;
        return $frame;
    }
1308726
public function readMessageFrame(string $subscriptionId, int $timeout): ?Frame
    {
        // pop up frame from the buffer
        if (isset($this->buffer[$subscriptionId]) && ($frame = array_shift($this->buffer[$subscriptionId]))) {
            --$this->currentBufferSize;

            return $frame;
        }

        // do nothing when buffer is full
        if ($this->currentBufferSize >= $this->bufferSize) {
            return null;
        }

        $startTime = microtime(true);
        $remainingTimeout = $timeout * 1000;

        while (true) {
            $this->getConnection()->setReadTimeout(0, $remainingTimeout);

            // there is nothing to read
            if (false === $frame = $this->readFrame()) {
                return null;
            }

            if ('MESSAGE' !== $frame->getCommand()) {
                throw new \LogicException(sprintf('Unexpected frame was received: "%s"', $frame->getCommand()));
            }

            $headers = $frame->getHeaders();

            if (false == isset($headers['subscription'])) {
                throw new \LogicException('Got message frame with missing subscription header');
            }

            // frame belongs to another subscription
            if ($headers['subscription'] !== $subscriptionId) {
                $this->buffer[$headers['subscription']][] = $frame;
                ++$this->currentBufferSize;

                $remainingTimeout -= (microtime(true) - $startTime) * 1000000;

                if ($remainingTimeout <= 0) {
                    return null;
                }

                continue;
            }

            return $frame;
        }
    }
1018229
function(params, callback) {
        if (!client) {
          fireEvent("error", {
            errorCode: 999,
            errorMessage: "ActiveMQ Client is not ready or has not been initialized!"
          });
        }

        var subscribeHeaders = {
          "destination": params.destination,
          "ack": params.ack
        };

        client.subscribe(subscribeHeaders, function(error, message) {
          if (error) {
            fireEvent("error", {
              errorCode: 999,
              errorMessage: "ActiveMQ Subscription Error!"
            });
            return;
          }

          message.readString("utf-8", function(error, body) {
            if (error) {
              fireEvent("error", {
                errorCode: 999,
                errorMessage: "ActiveMQ Read Message Error!"
              });
              return;
            }

            callback && callback(body);

            client.ack(message);
          });
        });
      }
12267
function persistSubscriptionLocally(subscription) {
    let subscriptionJSON = JSON.stringify(subscription);
    idb.open('web-push-db', 1).then(db => {
        let tx = db.transaction(['web-push-subcription'], 'readwrite');
        tx.objectStore('web-push-subcription').put({
            id: 1,
            data: subscriptionJSON
        });
        return tx.complete;
    });
}
182115
public function subscribe(Subscription $subscription)
    {
        if (!$this->isConnected) {
            return new RejectedPromise(new \LogicException('The client is not connected.'));
        }

        return $this->startFlow(new OutgoingSubscribeFlow([$subscription], $this->identifierGenerator));
    }
276628
@Override
    public java.util.concurrent.Future<GetSubscriptionAttributesResult> getSubscriptionAttributesAsync(String subscriptionArn) {

        return getSubscriptionAttributesAsync(new GetSubscriptionAttributesRequest().withSubscriptionArn(subscriptionArn));
    }
867673
public void createLocalDurableSubscription(
                                               ConsumerDispatcherState subState,
                                               TransactionCommon transaction)
                    throws
                    SIDurableSubscriptionAlreadyExistsException,
                    SIDiscriminatorSyntaxException,
                    SISelectorSyntaxException,
                    SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "createLocalDurableSubscription",
                        new Object[] { subState, transaction });

        // Create new getCursor and consumer dispatchers to return to the calling method
        ConsumerDispatcher consumerDispatcher = null;

        /*
         * Lock the list of durable subscriptions to prevent others creating/using an
         * identical subscription
         */
        synchronized (_consumerDispatchersDurable)
        {
            // If there's already a consumer dispatcher with this client name, then the
            // create must fail according to the current core API semantics.
            consumerDispatcher = getDurableSubscriptionConsumerDispatcher(subState);

            if (consumerDispatcher != null)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(
                               tc,
                               "createLocalDurableSubscription",
                               "Sub Already Exists - " + subState);

                throw new SIDurableSubscriptionAlreadyExistsException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_ALREADY_EXISTS_ERROR_CWSIP0143",
                                                        new Object[] {
                                                                      subState.getSubscriberID(),
                                                                      _messageProcessor.getMessagingEngineName() },
                                                        null));
            }
            /* We did not find the consumerDispatcher so we create it */

            /*
             * If a transaction was not provided...
             * Create a new LocalTransaction to perform the creation of a durable subscription.
             * Operations will include the creation of the itemstream,the persisting of
             * the durable state and the attaching of the consumerpoint.
             */
            boolean tranProvided = true;
            LocalTransaction siTran = null;

            if (transaction == null)
            {
                tranProvided = false;
                siTran =
                                _baseDestinationHandler
                                                .getTransactionManager()
                                                .createLocalTransaction(false);
                transaction = siTran;
            }

            /*
             * Create the consumerDispatcher, note we do not require this to be done under
             * the localTransaction. If something goes wrong after this operation, the
             * consumerDispatcher will be lost but thats ok.
             */

            Transaction msTran = _messageProcessor.resolveAndEnlistMsgStoreTransaction(transaction);

            try
            {
                /*
                 * Create a reference stream to be used by the durable subscription's
                 * consumer dispatcher
                 */
                final DurableSubscriptionItemStream durableSubscriptionReferenceStream =
                                new DurableSubscriptionItemStream(
                                                subState,
                                                _baseDestinationHandler.getDestinationManager(),
                                                _pubsubMessageItemStream,
                                                msTran);

                /* Create the consumer dispatcher to represent the subscription */
                consumerDispatcher =
                                configureDurableSubscription(durableSubscriptionReferenceStream);
                consumerDispatcher.registerControlAdapterAsMBean();

                //Venu mock mock
                // No ProxyHandler .. so commenting our this 
                /*
                 * // 166833.2
                 * // Generate the subscribeEvent for this new subscription
                 * // to see if this needs to be forwarded to a Neighbouring ME.
                 * _messageProcessor.getProxyHandler().subscribeEvent(
                 * subState,
                 * msTran);
                 */
            } catch (OutOfCacheSpace e)
            {
                // No FFDC code needed
                SibTr.exception(tc, e);
                // Rollback the transaction
                if (!tranProvided)
                    _baseDestinationHandler.handleRollback(siTran);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(
                               tc,
                               "createLocalDurableSubscription",
                               "SIResourceException");
                throw new SIResourceException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_CREATION_FAILED_CWSIP0033",
                                                        new Object[] {
                                                                      _baseDestinationHandler.getName(),
                                                                      subState.getSubscriberID(),
                                                                      e },
                                                        null),
                                e);

            } catch (MessageStoreException e)
            {
                // MessageStoreException shouldn't occur so FFDC.
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.createLocalDurableSubscription",
                                            "1:1093:1.35.2.4",
                                            this);
                SibTr.exception(tc, e);
                // Rollback the transaction
                if (!tranProvided)
                    _baseDestinationHandler.handleRollback(siTran);

                /*
                 * The message store has thrown an exception, which leaves us stuck. Record an ffst,
                 * write a trace record and throw a SIResourceException to the caller.
                 */
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createLocalDurableSubscription", e);

                SibTr.error(
                            tc,
                            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.createLocalDurableSubscription",
                                          "1:1112:1.35.2.4",
                                          e,
                                          _baseDestinationHandler.getName() });

                throw new SIResourceException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                      "1:1121:1.35.2.4",
                                                                      e,
                                                                      _baseDestinationHandler.getName() },
                                                        null),
                                e);
            } catch (SIDiscriminatorSyntaxException e)
            {
                // No FFDC code needed
                _baseDestinationHandler.handleRollback(siTran);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createLocalDurableSubscription", e);
                throw e;
            } catch (SISelectorSyntaxException e)
            {
                // No FFDC code needed
                _baseDestinationHandler.handleRollback(siTran);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createLocalDurableSubscription", e);
                throw e;
            }
            //Venu mock mock
            //commenting out as this exception became unreachable
            /*
             * catch (SIResourceException e)
             * {
             * // No FFDC code needed
             * _baseDestinationHandler.handleRollback(siTran);
             * if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
             * SibTr.exit(tc, "createLocalDurableSubscription", e);
             * throw e;
             * }
             */
            catch (RuntimeException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.createLocalDurableSubscription",
                                            "1:1153:1.35.2.4",
                                            this);

                if (siTran != null)
                    _baseDestinationHandler.handleRollback(siTran);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createLocalDurableSubscription", e);
                throw e;
            }

            if (!tranProvided)
            {
                /*
                 * If no transaction was provided, this call must have been made from
                 * the core api (i.e. JMS). We can not provide transaction at runtime.
                 * Since we are creating via JMS, we need to attach the consumer
                 * as part of the create operation.
                 */
                try
                {

                    /* We have our own transaction so commit it here */
                    siTran.commit();

                } catch (SIException e)
                {
                    // MessageStoreException shouldn't occur so FFDC.
                    FFDCFilter.processException(
                                                e,
                                                "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.createLocalDurableSubscription",
                                                "1:1182:1.35.2.4",
                                                this);

                    SibTr.exception(tc, e);
                    SibTr.error(
                                tc,
                                "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                new Object[] {
                                              "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                              "1:1191:1.35.2.4",
                                              e,
                                              _baseDestinationHandler.getName() });

                    // Failed to attach consumer, rollback the creation of the durable sub
                    // and throw exception.
                    _baseDestinationHandler.handleRollback(siTran);

                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "createLocalDurableSubscription", e);

                    throw new SIResourceException(
                                    nls.getFormattedMessage(
                                                            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                                            new Object[] {
                                                                          "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                          "1:1207:1.35.2.4",
                                                                          e,
                                                                          _baseDestinationHandler.getName() },
                                                            null),
                                    e);
                }
            }
            // Transaction was provided

        } // Release the durable subs lock

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createLocalDurableSubscription");
    }
1047191
protected void sendTransacted(Destination destination, Serializable message) {
        send(destination, message, null, null, true);
    }