/*
Convert the existing inbound JsMessage into a SubscriptionMessage.

Javadoc description supplied by JsMessage interface.
488842
public final SubscriptionMessage makeInboundSubscriptionMessage() throws IncorrectMessageTypeException {

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "makeInboundSubscriptionMessage");

    SubscriptionMessage subMessage = null;

    /* If the message has never been flattened/encoded, it will already be  */
    /* a SubscriptionMessage so we can just proceed with this recast.       */
    /* Otherwise we will need to convert it into a SubscriptionMessage.     */
    if (this instanceof SubscriptionMessageImpl) {
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Incoming message is already a SubscriptionMessage");
      subMessage = (SubscriptionMessageImpl)this;
    }
    /* The incoming message should be a SUBSCRIPTION message.                 */
    else if (getJsMessageType() == MessageType.SUBSCRIPTION) {
      // Ensure pending changes are flushed to JMF before passing the JMO around
      updateDataFields(MfpConstants.UDF_MAKE_INBOUND_OTHER);
      subMessage = new SubscriptionMessageImpl(jmo);
    }
    /* If the incoming message is some other sort of message it is an error.  */
    else {
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Not a SUBSCRIPTION Message");
      String exString = nls.getFormattedMessage("UNEXPECTED_MESSAGE_TYPE_CWSIF0102"
                                               ,new Object[] {getJsMessageType(), MessageType.SUBSCRIPTION}
                                               ,"The Message is not a SUBSCRIPTION Message");
      throw new IncorrectMessageTypeException(exString);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "makeInboundSubscriptionMessage");
    return subMessage;
  }
787969
public JsMessage transcribeToJmf() throws MessageCopyFailedException
                                          , IncorrectMessageTypeException
                                          , UnsupportedEncodingException {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "transcribeToJmf");
    JsMsgObject newJmo = null;

    // Try/catch block in case we get an MFPUnsupportedEncodingRuntimeException because the
    // payload is in an unsupported codepage.   d252277.2
    try {

     if (this instanceof JsJmsMessageImpl) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Transcribing from JsJmsMessageImpl");
        newJmo = jmo.transcribeToJmf();
      }

      else {
        // Non-JMS messages are not supported - they should have been converted into JMS by now
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, "Non-JMS Messages are not yet supported");
        String exString = nls.getFormattedMessage("UNEXPECTED_MESSAGE_TYPE_CWSIF0102"
                                                 ,new Object[] {getJsMessageType(), MessageType.JMS}
                                                 ,"The Message can not be represented as a pure JMF Message");
        throw new IncorrectMessageTypeException(exString);

      }
    }
    catch (MFPUnsupportedEncodingRuntimeException e) {
      FFDCFilter.processException(e, "com.ibm.ws.sib.mfp.impl.JsMessageImpl.transcribeToJmf", "721");
      // Throw the original UnsupportedEncodingException
      throw (UnsupportedEncodingException)e.getCause();
    }

    JsMessage newMsg;
    // if we have a new jmo, create a new message from it
    if (newJmo != this.jmo) {
      newMsg = createNewGeneralized(newJmo);
    }
    // otherwise, just return ourself
    else {
      newMsg = this;
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "transcribeToJmf", newMsg);
    return newMsg;
  }
1199418
protected SubscriptionMessage generateReplySubscriptionMessage()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "generateReplySubscriptionMessage");
      
    // Get the Message Handler for doing this operation.
    final SubscriptionMessageHandler messageHandler =
      iProxyHandler.getMessageHandler();
  
    // Reset the message.
    messageHandler.resetReplySubscriptionMessage();
      
    if (iLocalSubscriptions.size() > 0) 
    // Add the local subscriptions
      addToMessage(messageHandler, iLocalSubscriptions);
      
    if (iRemoteSubscriptions.size() > 0) 
      // Add the remote Subscriptions
      addToMessage(messageHandler, iRemoteSubscriptions);
        
    SubscriptionMessage message = messageHandler.getSubscriptionMessage();
   
    // Add the message back into the pool
    iProxyHandler.addMessageHandler(messageHandler);
      
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "generateReplySubscriptionMessage", message);
     
    return message;
  }
739206
static JsMessage createInboundWebMessage(String data) throws MessageDecodeFailedException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createInboundWebMessage");
        JsJmsMessage result = null;

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Decoding message from web data", data);

        // The inbound data consists of a number of '~' delimited fields. Refer to the
        // description in WebJsJmsMessageEncoderImpl for details.
        StringTokenizer st = new StringTokenizer(data, "~", true);

        // Pull out the header fields and the message payload.  Remember that it is possible
        // for fields to be empty strings which will be encoded consecutive delimiters.
        String version, id, type, format, topic, props, body;
        version = id = type = format = topic = props = body = null;
        try {
            version = st.nextToken();
            st.nextToken();
            String s = st.nextToken();
            if (!s.equals("~")) {
                id = s;
                st.nextToken();
            }
            s = st.nextToken();
            if (!s.equals("~")) {
                type = s;
                st.nextToken();
            }
            s = st.nextToken();
            if (!s.equals("~")) {
                format = s;
                st.nextToken();
            }
            s = st.nextToken();
            if (!s.equals("~")) {
                topic = s;
                st.nextToken();
            }
            s = st.nextToken();
            if (!s.equals("~")) {
                props = s;
                // This may be the end
                if (st.hasMoreTokens())
                    st.nextToken();
            }
            // The body may be omitted completely
            if (st.hasMoreTokens()) {
                body = st.nextToken();
            }
        } catch (NoSuchElementException e) {
            FFDCFilter.processException(e, "com.ibm.ws.sib.mfp.impl.WebJsMessageFactoryImpl.createInboundWebMessage", "119");
            throw new MessageDecodeFailedException("Inccorect WebClient encoding", e);
        }

        // Verify the data
        if (!version.equals("1"))
            throw new MessageDecodeFailedException("Incorrect WebClient encoding version: " + version);
        if (!format.startsWith(SIApiConstants.JMS_FORMAT))
            throw new MessageDecodeFailedException("Incorrect WebClient message format: " + format);

        // Decode the mesasge
        if (format.equalsIgnoreCase(SIApiConstants.JMS_FORMAT_TEXT))
            result = decodeTextBody(body);
        else if (format.equalsIgnoreCase(SIApiConstants.JMS_FORMAT_BYTES))
            result = decodeBytesBody(body);
        else if (format.equalsIgnoreCase(SIApiConstants.JMS_FORMAT_OBJECT))
            result = decodeObjectBody(body);
        else if (format.equalsIgnoreCase(SIApiConstants.JMS_FORMAT_STREAM))
            result = decodeStreamBody(body);
        else if (format.equalsIgnoreCase(SIApiConstants.JMS_FORMAT_MAP))
            result = decodeMapBody(body);
        else
            result = new JsJmsMessageImpl(MfpConstants.CONSTRUCTOR_NO_OP);

        decodeHeader(result, id, type, topic, props);

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Decoded Web Message", ((MessageImpl) result).debugMsg());
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createInboundWebMessage");
        return result;
    }
1373194
Observable<ChatResult> sendMessageWithAttachments(@NonNull final String conversationId, @NonNull final MessageToSend message, @Nullable final List<Attachment> attachments) {

        final MessageProcessor messageProcessor = attCon.createMessageProcessor(message, attachments, conversationId, getProfileId());

        return checkState()
                .flatMap(client ->
                {
                    messageProcessor.preparePreUpload(); // convert fom too large message parts to attachments, adds temp upload parts for all attachments
                    return upsertTempMessage(messageProcessor.createTempMessage()) // create temporary message
                            .flatMap(isOk -> attCon.uploadAttachments(messageProcessor.getAttachments(), client)) // upload attachments
                            .flatMap(uploaded -> {
                                if (uploaded != null && !uploaded.isEmpty()) {
                                    messageProcessor.preparePostUpload(uploaded); // remove temp upload parts, add parts with upload data
                                    return upsertTempMessage(messageProcessor.createTempMessage()); // update message with attachments details like url
                                } else {
                                    return Observable.fromCallable(() -> true);
                                }
                            })
                            .flatMap(isOk -> client.service().messaging().sendMessage(conversationId, messageProcessor.prepareMessageToSend()) // send message with attachments details as additional message parts
                                    .flatMap(result -> result.isSuccessful() ? updateStoreWithSentMsg(messageProcessor, result) : handleMessageError(messageProcessor, new ComapiException(result.getErrorBody()))) // update temporary message with a new message id obtained from the response
                                    .onErrorResumeNext(t -> handleMessageError(messageProcessor, t))); // if error occurred update message status list adding error status
                });
    }
356046
@Override
    public Message createMessage(Session session) throws JMSException {
        Message message = session.createTextMessage(requestMessage);
        if (correlationId != null)
            message.setJMSCorrelationID(correlationId);
        if (replyQueue != null) {
            message.setJMSReplyTo(replyQueue);
        }
        if (delaySeconds >0) {
            ApplicationContext.getJmsProvider().setMessageDelay(null, message, delaySeconds);
        }
        return message;
    }
397337
private SubscriptionMessage createSubscriptionMessage()
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "createSubscriptionMessage");

    ControlMessageFactory factory = null;
    SubscriptionMessage subscriptionMessage = null;

    try
    {
      factory = MessageProcessor.getControlMessageFactory();
      subscriptionMessage = factory.createNewSubscriptionMessage();
    }
    catch (Exception e)
    {
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.proxyhandler.SubscriptionMessageHandler.createSubscriptionMessage",
        "1:162:1.34",
        this);
        
      SibTr.exception(tc, e);
    }

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "createSubscriptionMessage", subscriptionMessage);

    return subscriptionMessage;
  }
725011
void addNewMessage(JsMessageWrapper message, boolean isStored, boolean isRecoverable)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc,
                  "addNewMessage",
                  new Object[] { new Integer(hashCode()),
                                 message,
                                 new Boolean(isStored)});

    JsMessage jsMsg = message.getMessage();

    long id = jsMsg.getSystemMessageValue();
    SIBUuid8 uuid = jsMsg.getSystemMessageSourceUuid();

    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
      SibTr.debug(tc,
                  "addNewMessage",
                  new Object[] { new Integer(hashCode()),
                                 new Long(jsMsg.getSystemMessageValue()),
                                 jsMsg.getSystemMessageSourceUuid(),
                                 message,
                                 new Boolean(isStored)});

    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
      SibTr.debug(localConsumerPoint, tc, "verboseMsg OUT : " + message.getMessage().toVerboseString());

    synchronized(this)
    {
      LMEMessage newElement = null;
      long messageReferenceExpiry = messageProcessor.getCustomProperties().get_message_reference_expiry_value();

      // We have to create an object if the pool is empty
      if(pooledMsg == null)
      {
        newElement = new LMEMessage(id, uuid, message, isStored, isRecoverable, messageLockExpiry, messageReferenceExpiry);
      }
      // Otherwise pop the first pooled object and use that
      else
      {
        newElement = pooledMsg;
        pooledMsg = pooledMsg.next;
        newElement.next = null;
        pooledCount--;

        newElement.reuseMessage(id, uuid, message, isStored, isRecoverable, messageLockExpiry, messageReferenceExpiry);
      }

      // Add the message to the end of the list
      newElement.previous = lastMsg;
      if(lastMsg != null)
        lastMsg.next = newElement;
      else
        firstMsg = newElement;
      lastMsg = newElement;

      // If the locks are set to expire and this is the first message we need to
      // register an alarm
      if(messageLockExpiry != 0)
      {
        // If there are no other messages expiring, this will be the first one
        if(nextMsgToExpire == null)
          nextMsgToExpire = newElement;

        // If we don't have an alarm registered, register one now
        if(!alarmRegistered)
        {
          if(TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "addNewMessage", "Registering MessageLock Expiry alarm for: " + nextMsgToExpire);

          alarmManager.create(messageLockExpiry, this);
          alarmRegistered = true;
        }
      }

      if (messageReferenceExpiry != 0)
      {
        if (messageLockExpiry != 0 && messageReferenceExpiry > messageLockExpiry)
        {
          // The messageReferenceExpiry is greater than the messageLockExpiry. There is no
          // point in creating the alarm for messageReferenceExpiry as the message lock expiry
          // will always go first.
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          {
            SibTr.debug(tc, "MessageReferenceExpiry: "+messageReferenceExpiry+" is greater than messageLockExpiry: "+messageLockExpiry);
            SibTr.debug(tc, "MessageReferneceExpiry Alarm not registered");
          }
        }
        else
        {
          try
          {
            if (message.getReportCOD() == null)
            {
              // Now register the message reference expiry alarm
              // If there are no other message references expiring, this will be the first one
              if(nextMsgReferenceToExpire == null)
                nextMsgReferenceToExpire = newElement;

              // If we don't have an alarm registered, register one now
              if(!msgReferenceAlarmRegistered)
              {
                if(TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                  SibTr.debug(tc, "addNewMessage", "Registering MessageReference Expiry alarm for: " + nextMsgReferenceToExpire);

                alarmManager.create(messageReferenceExpiry, new MessageReferenceExpiryAlarm());
                msgReferenceAlarmRegistered = true;
              }
            }
          }
          catch(SIResourceException e)
          {
            // No FFDC code needed

            // There was a problem getting hold of the ReportCOD of the message
            // assume it is set and don't create the expiry alarm.
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
              SibTr.debug(tc, "Thrown exception when trying to create msg ref expiry alarm: " + e);
          }
        }
     }

      // By adding a message we must be about to call consumeMessages and therefore
      // we're in a valid state for calls
      validState = true;
    } // synchronized

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "addNewMessage", this);
  }
884854
@Override
	public Promise subscribe(String channel) {
		try {

			// Create publisher
			createOrGetPublisher(channel);

			// Create subscriber
			synchronized (subscribers) {
				if (!subscribers.containsKey(channel)) {
					TopicSubscriber subscriber = session.createSubscriber(session.createTopic(channel));
					subscribers.put(channel, subscriber);
					subscriber.setMessageListener((message) -> {
						try {
							BytesMessage msg = (BytesMessage) message;
							byte[] bytes = new byte[(int) msg.getBodyLength()];
							msg.readBytes(bytes);
							received(channel, bytes);
						} catch (Exception cause) {
							logger.error("Unable to deserialize message!", cause);
						}
					});
				}
			}
		} catch (Exception cause) {
			return Promise.reject(cause);
		}
		return Promise.resolve();
	}
806485
@Override
    public TextMessage createTextMessage() throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createTextMessage");

        TextMessage textMessage = null;

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

        return textMessage;

    }
320541
public void unsubscribe( String clientID , String subscriptionName ) throws JMSException
    {
    	String subscriberID = clientID+"-"+subscriptionName;
    	
    	subscriptionsLock.writeLock().lock();
    	try
		{
        	LocalTopicSubscription subscription = subscriptionMap.get(subscriberID);
        	if (subscription == null)
        		return;
        	
        	if (isConsumerRegistered(subscriberID))
        		throw new FFMQException("Subscription "+subscriptionName+" is still in use","SUBSCRIPTION_STILL_IN_USE");
    		
        	subscriptionMap.remove(subscriberID);
        	removeSubscription(subscription);
		}
    	finally
    	{
    		subscriptionsLock.writeLock().unlock();
    	}
    }
362234
public static Builder newBuilder(ProjectSubscriptionName subscription, MessageReceiver receiver) {
    return newBuilder(subscription.toString(), receiver);
  }
1065328
private void send(Message message, String destinationName, TestContext context) {
        if (log.isDebugEnabled()) {
            log.debug("Sending JMS message to destination: '" + destinationName + "'");
        }

        endpointConfiguration.getJmsTemplate().send(destinationName, session -> {
            javax.jms.Message jmsMessage = endpointConfiguration.getMessageConverter().createJmsMessage(message, session, endpointConfiguration, context);
            endpointConfiguration.getMessageConverter().convertOutbound(jmsMessage, message, endpointConfiguration, context);
            return jmsMessage;
        });

        log.info("Message was sent to JMS destination: '" + destinationName + "'");
    }
293951
@Override
	public TopicSubscriber createSubscriber(Topic topic) throws JMSException
    {
        return createSubscriber(topic,null,false);
    }
59206
@Override
    public BytesMessage createBytesMessage() throws JMSRuntimeException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createBytesMessage");

        BytesMessage bytesMessage = null;

        try {
            bytesMessage = jmsSession.createBytesMessage();
        } catch (JMSException e) {
            throw (JMSRuntimeException) JmsErrorUtils.getJMS2Exception(e, JMSRuntimeException.class);
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "createBytesMessage");
        }
        return bytesMessage;
    }
718060
public static Builder from(ServiceMessage message) {
    return ServiceMessage.builder().data(message.data()).headers(message.headers());
  }
392646
function subscribeToResult(outerSubscriber, result, outerValue, outerIndex) {
    var destination = new _InnerSubscriber__WEBPACK_IMPORTED_MODULE_0__["InnerSubscriber"](outerSubscriber, outerValue, outerIndex);
    return Object(_subscribeTo__WEBPACK_IMPORTED_MODULE_1__["subscribeTo"])(result)(destination);
}