Creates a new {@link QueueConsumer}. For every text message received, the callback
is invoked with the contents of the text message as string.
763123
public QueueConsumer createQueueTextMessageListener(final String topic, final ConsumerCallback<String> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new QueueConsumer(connectionFactory, jmsConfig, topic, new TextMessageConsumerCallback(messageCallback));
    }
412718
public TopicConsumer createTopicTextMessageListener(final String topic, final ConsumerCallback<String> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicConsumer(connectionFactory, jmsConfig, topic, new TextMessageConsumerCallback(messageCallback));
    }
841738
@Override
    public Message newMessage(final String message) {
        final ReusableSimpleMessage result = getSimple();
        result.set(message);
        return result;
    }
1342494
@Override
    public JmsQueue createQueue(String name) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createQueue", name);
        JmsQueue queue = null;

        // if name string is null, empty or just "queue://", throw exception
        if ((name == null) || ("".equals(name)) || (JmsQueueImpl.QUEUE_PREFIX.equals(name))) {
            throw (InvalidDestinationException) JmsErrorUtils.newThrowable(
                                                                           InvalidDestinationException.class,
                                                                           "INVALID_VALUE_CWSIA0003",
                                                                           new Object[] { "Queue name", name },
                                                                           tc
                            );
        }

        // if name is "topic://" throw exception
        if (name.startsWith(JmsTopicImpl.TOPIC_PREFIX)) {
            throw (InvalidDestinationException) JmsErrorUtils.newThrowable(
                                                                           InvalidDestinationException.class,
                                                                           "MALFORMED_DESCRIPTOR_CWSIA0047",
                                                                           new Object[] { "Queue", name },
                                                                           tc
                            );
        }

        name = name.trim();
        queue = (JmsQueue) destCreator.createDestinationFromString(name, URIDestinationCreator.DestType.QUEUE);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createQueue", queue);
        return queue;
    }
544279
private MessageConsumer createConsumer(Destination destination, String name, String messageSelector, boolean noLocal,
                                           boolean isSharedNonDurable) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createConsumer", new Object[] { destination, messageSelector, noLocal, isSharedNonDurable });
        JmsMsgConsumer consumer = null;

        checkNotClosed();
        if (destination == null) {
            throw (InvalidDestinationException) JmsErrorUtils.newThrowable(InvalidDestinationException.class,
                                                                           "INVALID_VALUE_CWSIA0048",
                                                                           new Object[] { "Destination", "null" },
                                                                           tc
                            );
        }

        if (isSharedNonDurable) {
            if ((name == null) || ("".equals(name))) {
                throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                                "INVALID_VALUE_CWSIA0048",
                                                                new Object[] { "sharedSubscriptionName", name },
                                                                tc
                                );
            }
        }

        // Check that this is not a foreign implementation.
        // Note that in some situations this method call can return a _different_
        // destination than the one passed in as the parameter (e.g. Spring).
        JmsDestinationImpl nativeDest = JmsDestinationImpl.checkNativeInstance(destination);

        // Check that this destination is not blocked from creating consumers
        JmsDestinationImpl.checkBlockedStatus(nativeDest);

        // create the consumer properties object filling in as much information
        // as we have available at the moment.
        ConsumerProperties newProps = new ConsumerProperties(nativeDest,
                        null, // DestType                filled in later
                        messageSelector, // Selector
                        null, // Reliability             filled in later
                        false, // ReadAhead               filled in later
                        true, // RecovExpress            filled in later
                        noLocal, // NoLocal
                        name, // SubName  .. with JMS 2.0, the value is set for all cases
                        connection.getClientID(), // ClientID .. with JMS 2.0, the value is set for all cases
                        isSharedNonDurable, // SupportMultiple .. with JMS 2.0, the value is set for all cases         
                        null // DurableSubscriptionHome n/a for non-durable consumers
        );

        // Hold the state lock while the msg consumer constructor is called to prevent
        // it from reading invalid start/stopped state while start is in progress. This
        // will basically cause and createConsumer calls on this session to wait until
        // session.start is complete.
        synchronized (stateLock) {
            consumer = instantiateConsumer(newProps);
            // newly created consumers must be synchronous
            syncConsumers.add(consumer);

            // d353701 - output a warning message if there are 'lots' of producers active.
            int totalConsumers = syncConsumers.size() + asyncConsumers.size();
            if (totalConsumers % PROD_CONS_WARNING_THRESHOLD == 0) {
                // We wish to tell the user which line of their application created the session,
                // so we must obtain a line of stack trace from their application.
                String errorLocation = JmsErrorUtils.getFirstApplicationStackString();
                SibTr.warning(tc, "MANY_CONSUMERS_WARNING_CWSIA0059", new Object[] { "" + totalConsumers, errorLocation });
            }

        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createConsumer", consumer);
        return consumer;
    }
113127
@SuppressWarnings("unchecked")
    public static UniversalIdIntMessage newInstance() {
        Date now = new Date();
        UniversalIdIntMessage msg = new UniversalIdIntMessage();
        msg.setId(QueueUtils.IDGEN.generateId64()).setTimestamp(now);
        return msg;
    }
1215891
protected IQueueMessage<ID, DATA> takeFromQueue() {
        KafkaMessage kMsg = kafkaClient.consumeMessage(consumerGroupId, true, topicName, 1000,
                TimeUnit.MILLISECONDS);
        return kMsg != null ? deserialize(kMsg.content()) : null;
    }
1279036
public static QueuedMessage createSIBQueuedMessage(SIMPQueuedMessageControllable qmc) {

        String id = null;
        int jsApproximateLength = 0;
        String name = null;
        String state = null;
        String transactionId = null;
        String type = null;
        String busSystemMessageId = null;

        id = qmc.getId();
        name = qmc.getName();
        state = null;
        transactionId = null;

        try {
            if (qmc.getState() != null) {
                state = qmc.getState().toString();
            }
            transactionId = qmc.getTransactionId();
        } catch (SIMPException e) {
            // No FFDC code needed
        }

        try {
            JsMessage jsMessage = qmc.getJsMessage();
            jsApproximateLength = jsMessage.getApproximateLength();
            busSystemMessageId = jsMessage.getSystemMessageId();
            type = jsMessage.getJsMessageType().toString();
        } catch (SIMPControllableNotFoundException e) {
            // No FFDC code needed
        } catch (SIMPException e) {
            // No FFDC code needed
        }

        return new QueuedMessage(id, name, jsApproximateLength, state, transactionId, type, busSystemMessageId);
    }
718627
public RemoteReceiveQueue createRemoteReceiveQueue(String strQueueName, String strQueueType) throws RemoteException
    {
        MessageManager messageManager = this.getEnvironment().getMessageManager(this.getApplication(), true);
        BaseMessageReceiver receiver = (BaseMessageReceiver)messageManager.getMessageQueue(strQueueName, strQueueType).getMessageReceiver();
        ReceiveQueueSession remoteQueue = new ReceiveQueueSession(this, receiver);
        return remoteQueue;
    }
1270369
public static QueuedMessageDetail createSIBQueuedMessageDetail(
                                                                   SIMPQueuedMessageControllable qmc, Locale locale) {

        String type = null;
        boolean fullyValid = true;
        boolean valid = false;

        // Data members of SIBusMessage
        String busDiscriminator = null;
        Integer busPriority = null;
        String busReliability = null;
        Long busTimeToLive = null;
        String busReplyDiscriminator = null;
        Integer busReplyPriority = null;
        String busReplyReliability = null;
        Long busReplyTimeToLive = null;
        String busSystemMessageId = null;
        Long busExceptionTimestamp = null;
        String busExceptionMessage = null;
        // F001333-14609
        String busExceptionProblemSubscription = null;
        String busExceptionProblemDestination = null;

        // Data members of JsMessage
        String jsMessageType = null;
        int jsApproximateLength = -1;
        Long jsTimestamp = null;
        Long jsMessageWaitTime = null;
        Long jsCurrentMEArrivalTimestamp = null;
        Integer jsRedeliveredCount = null;
        String jsSecurityUserid = null;
        String jsProducerType = null;
        byte[] jsApiMessageIdAsBytes = null;
        byte[] jsCorrelationIdAsBytes = null;

        // Data members of JsApiMessage
        String apiMessageId = null;
        String apiCorrelationId = null;
        String apiUserid = null;
        String apiFormat = null;

        // Data members of JsJmsMessage
        String jmsDeliveryMode = null;
        Long jmsExpiration = null;
        String jmsDestination = null;
        String jmsReplyTo = null;
        Boolean jmsRedelivered = null;
        String jmsType = null;
        int jmsXDeliveryCount = -1;
        String jmsXAppId = null;

        // Inherited attributes
        String id = null;
        String name = null;
        String state = null;
        long timestamp = 0; // Never set!!!
        String transactionId = null;

        JsMessage jsMessage = null; // JsMesaage is a subclass of SIBusMessage

        try {
            jsMessage = qmc.getJsMessage();
            valid = true;
        } catch (SIMPControllableNotFoundException e) {
            // No FFDC code needed
            fullyValid = false;
        } catch (SIMPException e) {
            // No FFDC code needed
            fullyValid = false;
        }

        id = qmc.getId();
        if ("-1".equals(id)) {
            fullyValid = false;
        }
        name = qmc.getName();

        try {
            if (qmc.getState() != null) {
                state = qmc.getState().toString();
            }
            transactionId = qmc.getTransactionId();
        } catch (SIMPException e) {
            // No FFDC code needed
            fullyValid = false;
        }

        busDiscriminator = jsMessage.getDiscriminator();
        busPriority = jsMessage.getPriority();
        busReliability = jsMessage.getReliability().toString();
        busTimeToLive = jsMessage.getTimeToLive();
        busReplyDiscriminator = jsMessage.getReplyDiscriminator();
        busReplyPriority = jsMessage.getReplyPriority();

        Reliability replyReliability = jsMessage.getReplyReliability();

        if (replyReliability != null) {
            busReplyReliability = replyReliability.toString();
        }

        busReplyTimeToLive = jsMessage.getReplyTimeToLive();
        busSystemMessageId = jsMessage.getSystemMessageId();

        Integer exceptionReason = jsMessage.getExceptionReason();

        if (exceptionReason != null) {

            busExceptionTimestamp = jsMessage.getExceptionTimestamp();

            if (locale == null) {
                locale = Locale.getDefault();
            }

            busExceptionMessage = TraceNLS.getFormattedMessage(JsAdminConstants.EXCEPTION_MSG_BUNDLE,
                                                               JsAdminConstants.EXCEPTION_MESSAGE_KEY_PREFIX + exceptionReason, locale, jsMessage
                                                                               .getExceptionInserts(), "Exception Message");

            // F001333-14609
            busExceptionProblemSubscription = jsMessage.getExceptionProblemSubscription();

            busExceptionProblemDestination = jsMessage.getExceptionProblemDestination();
        }

        jsMessageType = jsMessage.getJsMessageType().toString();
        jsApproximateLength = jsMessage.getApproximateLength();
        jsTimestamp = jsMessage.getTimestamp();
        jsMessageWaitTime = jsMessage.getMessageWaitTime();
        jsCurrentMEArrivalTimestamp = jsMessage.getCurrentMEArrivalTimestamp();
        jsRedeliveredCount = jsMessage.getRedeliveredCount();
        jsSecurityUserid = jsMessage.getSecurityUserid();
        jsProducerType = jsMessage.getProducerType().toString();
        jsApiMessageIdAsBytes = jsMessage.getApiMessageIdAsBytes();
        jsCorrelationIdAsBytes = jsMessage.getCorrelationIdAsBytes();

        if (jsMessage instanceof JsApiMessage) {
            JsApiMessage jsApiMessage = (JsApiMessage) jsMessage;
            apiMessageId = jsApiMessage.getApiMessageId();
            apiCorrelationId = jsApiMessage.getCorrelationId();
            apiUserid = jsApiMessage.getUserid();
            apiFormat = jsApiMessage.getFormat();
        }

        if (jsMessage instanceof JsJmsMessage) {

            JsJmsMessage jsJmsMessage = (JsJmsMessage) jsMessage;

            jmsDeliveryMode = jsJmsMessage.getJmsDeliveryMode().toString();
            jmsExpiration = jsJmsMessage.getJmsExpiration();
            type = "JMS";

            /*
             * if (jsJmsMessage.getJmsDestination() != null) {
             * try {
             * JmsDestination dest = JmsInternalsFactory.getMessageDestEncodingUtils()
             * .getDestinationFromMsgRepresentation(jsJmsMessage.getJmsDestination());
             * jmsDestination = dest.toString();
             * }
             * catch (Exception e) {
             * // No FFDC code needed
             * fullyValid = false;
             * }
             * }
             */

            // Get JMS reply to destination from full header JMS message.

            /*
             * try {
             * Message fullJMSMsg = JmsInternalsFactory.getSharedUtils().inboundMessagePath(jsJmsMessage,
             * null, null); // Added 'null' parameter for pass thru props (SIB0121)
             * Destination dest = fullJMSMsg.getJMSReplyTo();
             * 
             * if (dest != null) {
             * jmsReplyTo = dest.toString();
             * }
             * }
             * catch (Exception e) {
             * // No FFDC code needed
             * fullyValid = false;
             * }
             */

            jmsRedelivered = jsJmsMessage.getJmsRedelivered();
            jmsType = jsJmsMessage.getJmsType();
            jmsXDeliveryCount = jsJmsMessage.getJmsxDeliveryCount();
            jmsXAppId = jsJmsMessage.getJmsxAppId();
        }

        return new QueuedMessageDetail(id, name, state, timestamp, transactionId, type,
                        busDiscriminator, busPriority, busReliability, busTimeToLive, busReplyDiscriminator,
                        busReplyPriority, busReplyReliability, busReplyTimeToLive, busSystemMessageId,
                        busExceptionTimestamp, busExceptionMessage, busExceptionProblemSubscription,
                        busExceptionProblemDestination, jsMessageType, jsApproximateLength,
                        jsTimestamp, jsMessageWaitTime, jsCurrentMEArrivalTimestamp, jsRedeliveredCount,
                        jsSecurityUserid, jsProducerType, jsApiMessageIdAsBytes, jsCorrelationIdAsBytes,
                        apiMessageId, apiCorrelationId, apiUserid, apiFormat, jmsDeliveryMode, jmsExpiration,
                        jmsDestination, jmsReplyTo, jmsRedelivered, jmsType, jmsXDeliveryCount, jmsXAppId,
                        fullyValid, valid);
    }
597021
public static CompletableFuture<IMessageSender> createMessageSenderFromConnectionStringBuilderAsync(ConnectionStringBuilder amqpConnectionStringBuilder) {
        Utils.assertNonNull("amqpConnectionStringBuilder", amqpConnectionStringBuilder);
        return createMessageSenderFromEntityPathAsync(amqpConnectionStringBuilder.getEndpoint(), amqpConnectionStringBuilder.getEntityPath(),  Util.getClientSettingsFromConnectionStringBuilder(amqpConnectionStringBuilder));
    }
1241353
@Override
    public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool,
            int maxMessages) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }
554091
def consume(self, on_message, queue):
        
        self.channel.queue_declare(queue=queue, durable=True, exclusive=False,
                                   auto_delete=False)
        self.channel.queue_bind(queue=queue, exchange=self._exchange,
                                routing_key=self._routing_key)
        self.channel.basic_qos(prefetch_count=1)
        # Setup up our consumer callback
        self.channel.basic_consume(on_message, queue)
        # This is blocking until channel.stop_consuming is called
        # and will allow us to receive messages
        self.channel.start_consuming()
431254
private synchronized void createConsumerThreadPool() {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "createConsumerThreadPool");
        if (_consumerThreadPool == null) {
            // 214163 Add a threadpool listener to associate threads in trace
            // with the messaging engine.
            _consumerThreadPool = new ThreadPool(
                            "Consumer " + getMessagingEngineName(),
                            0,
                            _customProperties.get_max_consumer_threadpool_size(),
                            new ThreadPoolListener[] { new ThreadPoolListenerImpl(this) });
            _consumerThreadPool
                            .setRequestBufferExpansionLimit(Integer.MAX_VALUE);
        }
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "createConsumerThreadPool");
    }
1298050
@Override
    public Queue createQueue(String queueName) throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createQueue", new Object[] { queueName });

        Queue queue = null;
        try {
            queue = jmsSession.createQueue(queueName);
        } catch (JMSException jmse) {
            throw (JMSRuntimeException) JmsErrorUtils.getJMS2Exception(jmse, JMSRuntimeException.class);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "createQueue", new Object[] { queue });
        }

        return queue;
    }
139814
@Override
    public MessageConsumer createDurableConsumer(Topic topic,
                                                 String name)
                    throws IllegalStateException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createDurableConsumer", new Object[] { topic, name });
        throw (javax.jms.IllegalStateException) JmsErrorUtils.newThrowable(javax.jms.IllegalStateException.class,
                                                                           "INVALID_OP_FOR_CLASS_CWSIA0481",
                                                                           new Object[] { "createDurableConsumer", "QueueSession" },
                                                                           tc);

    }
913649
public ReceiverQueue createReceiver(int limit) {
		synchronized (receivers) {
			ReceiverQueue q = null;
			if (!closed && (maxNrofReceivers == 0 || receivers.size() < maxNrofReceivers)) {
				q = new ReceiverQueue(/*this, */limit);
				receivers.add(q);
			}
			return q;
		}
	}