Method sendCODMessage.
Initializes the reportHandler and sends a COD message if appropriate

@param msg
@param transaction
14881
void sendCODMessage(SIMPMessage msg, TransactionCommon tran) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendCODMessage", new Object[] { msg, tran });

        // If COD Report messages are required, this is when we need to create and
        // send them.

        if (msg.getReportCOD() != null)
        {
            // Create the ReportHandler object if not already created
            if (_reportHandler == null)
                _reportHandler = new ReportHandler(messageProcessor);

            try
            {
                _reportHandler.handleMessage(msg, tran, SIApiConstants.REPORT_COD);
            } catch (SIException e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.BaseDestinationHandler.sendCODMessage",
                                            "1:3909:1.700.3.45",
                                            this);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "sendCODMessage", "SIResourceException");
                throw new SIResourceException(
                                nls.getFormattedMessage(
                                                        "REPORT_MESSAGE_ERROR_CWSIP0423",
                                                        new Object[] { messageProcessor.getMessagingEngineName(), e },
                                                        null),
                                e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendCODMessage");
    }
639399
public void send(final Command msg) throws HandlerSocketException {
		Session session = this.selectSession();
		session.write(msg);
	}
440123
@Override
    public void send(Message msg) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "send", msg);

        //Defer straight through to the version with completion listner (which will do all the locking and checking).
        send_internal(msg, null);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "send");
    }
560365
public static void insertMessage(Throwable onObject, String msg) {
        try {
            Field field = Throwable.class.getDeclaredField("detailMessage"); //Method("initCause", new Class[]{Throwable.class});
            field.setAccessible(true);
            if (onObject.getMessage() != null) {
                field.set(onObject, "\n[\n" + msg + "\n]\n[\nMessage: " + onObject.getMessage() + "\n]");
            } else {
                field.set(onObject, "\n[\n" + msg + "]\n");
            }
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
        }
    }
432386
public void send(String channel, String message) {
		// CAUSE: Assignment to method parameter
		Pair<Boolean, String> sendValidation = isSendValid(channel, message, null, null);

		//String lMessage = message.replace("\n", "\\n");

		if (sendValidation != null && sendValidation.first) {
			try {
				String messageId = Strings.randomString(8);
				ArrayList<Pair<String, String>> messagesToSend = multiPartMessage(
						message, messageId);
				for (Pair<String, String> messageToSend : messagesToSend) {
					send(channel, messageToSend.second, messageToSend.first,
							sendValidation.second);
				}
			} catch (IOException e) {
				raiseOrtcEvent(EventEnum.OnException, this, e);
			}
		}
	}
1178840
public void sendBasicMessageAsync(Session session, BasicMessage msg) {
        String text = ApiDeserializer.toHawkularFormat(msg);
        sendTextAsync(session, text);
    }
845033
@Override
    public void eventPostCommitAdd(Transaction transaction) throws SevereMessageStoreException
    {
        super.eventPostCommitAdd(transaction);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "eventPostCommitAdd",
                        new Object[] { transaction });

        if ((getMessageItem().getRegisterForPostEvents()) || failedInitInRestore)
        {
            stream = (MessageReferenceStream) getReferenceStream();
            //we can only register the listeners if we are fully initialized
            if ((stream instanceof DurableSubscriptionItemStream) || (stream instanceof ProxyReferenceStream))
            {
                stream.registerListeners(this);
            }
            failedInitInRestore = false;
            getMessageItem().setRegisterForPostEvents(failedInitInRestore);

        }

        if (!transaction.isAutoCommit())
        {
            try
            {
                if (POST_COMMIT_ADD_1 != null)
                {
                    POST_COMMIT_ADD_1.messageEventOccurred(MessageEvents.POST_COMMIT_ADD, this, transaction);
                }
                if (POST_COMMIT_ADD_2 != null)
                {
                    POST_COMMIT_ADD_2.messageEventOccurred(MessageEvents.POST_COMMIT_ADD, this, transaction);
                }
            } catch (SIException e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.store.items.MessageItemReference.eventPostCommitAdd",
                                            "1:422:1.147",
                                            this);

                SibTr.exception(tc, e);

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

                throw new SIErrorException(e);
            }
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "eventPostCommitAdd");
    }
223893
private void eventMessageExpiryNotification(
                                                SIMPMessage msg,
                                                TransactionCommon tran) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "eventMessageExpiryNotification",
                        new Object[] { msg, tran });

        // If a ReportHandler object has not been created yet, then do so.
        if (_reportHandler == null)
            _reportHandler = new ReportHandler(messageProcessor);

        // Generate and send the report under the same transaction as the delete
        try
        {
            _reportHandler.handleMessage(msg, tran, SIApiConstants.REPORT_EXPIRY);
        } catch (SIException e)
        {
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.BaseDestinationHandler.eventMessageExpiryNotification",
                                        "1:3778:1.700.3.45",
                                        this);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "eventMessageExpiryNotification", "SIResourceException");
            throw new SIResourceException(
                            nls.getFormattedMessage(
                                                    "REPORT_MESSAGE_ERROR_CWSIP0422",
                                                    new Object[] { messageProcessor.getMessagingEngineName(), e },
                                                    null),
                            e);
        }

        /**
         * 463584
         * If this message has been restored from msgstore but this destination has been recreated
         * in admin thus a different destination uuid and diferent BDH instance the stats object will be null.
         * It only gets initialized during the create/updateLocalisations call. This call will never happen
         * on this BDH.
         * Therefore... If a destination has been recreated, the stats on expiry will no longer be valid.
         */
        if (_protoRealization != null)
            _protoRealization.onExpiryReport();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "eventMessageExpiryNotification");
    }
882618
@Override
    public void deliverExpressMessage(MessageItem msgItem,
                                      ExpressTargetStream expressTargetStream)

                    throws SIIncorrectCallException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "deliverExpressMessage", new Object[] { msgItem });

        //Create transaction
        LocalTransaction siTran = _txManager.createLocalTransaction(false);

        try
        {
            long endTick = msgItem.getMessage().getGuaranteedValueValueTick();

            JsDestinationAddress routingDestinationAddr =
                            msgItem.getMessage().getRoutingDestination();
            if (routingDestinationAddr != null)
            {
                DestinationHandler routingDestination = null;
                try
                {
                    routingDestination =
                                    _destinationManager.getDestination(routingDestinationAddr, false, true);

                    if (_isLink)
                    {
                        LinkHandler link = (LinkHandler) _destination;

                        // If this is coming in off a link then use the link's configuration
                        // to determine if any local message points should be preferred
                        msgItem.setPreferLocal(link.preferLocalTargetQueuePoint());

                        String linkInboundUserid = ((LinkHandler) _destination).getInboundUserid();
                        if (linkInboundUserid != null)
                        {
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                SibTr.debug(tc, "Set link inbound userid: " + linkInboundUserid + ", into message");
                            // Call SIB.security (ultimately) to set inbounduserid into msg
                            _messageProcessor.getAccessChecker().setSecurityIDInMessage(
                                                                                        linkInboundUserid,
                                                                                        msgItem.getMessage());
                        }
                    }
                    // If it's come from another ME in the bus then they're sending it
                    // to this message point because they want to so we prefer the local one.
                    // This allows the mesasge to be re-routed if absolutely necessary.
                    else
                        msgItem.setPreferLocal(true);

                    // Call the handler to deal with the message - passing null as the producer
                    // session id.
                    ProducerInputHandler inputHandler =
                                    (ProducerInputHandler) routingDestination.getInputHandler();

                    inputHandler.handleMessage(msgItem,
                                               siTran,
                                               _messageProcessor.getMessagingEngineUuid());

                } catch (SIException e)
                {
                    // No FFDC code needed

                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    {
                        SibTr.exception(tc, e);
                        SibTr.debug(tc, "Discarding the express message as it could not be from the targetStreamManager");
                    }
                }
            }
            else
            {
                ConsumerDispatcher consumerDispatcher;

                consumerDispatcher = (ConsumerDispatcher) _destination.getLocalPtoPConsumerManager();
                consumerDispatcher.put(msgItem, siTran, null, false);
            }

            // Update the completedPrefix on the stream
            // This will put all ticks before this point into Completed state
            expressTargetStream.setCompletedPrefix(endTick);

            siTran.commit();
        } catch (SIIncorrectCallException e)
        {
            // No FFDC code needed
            handleRollback(siTran);

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

            throw e;
        } catch (SIResourceException e)
        {
            // No FFDC code needed
            handleRollback(siTran);

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

            throw e;
        } catch (RuntimeException e)
        {
            // FFDC
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.PtoPInputHandler.deliverExpressMessage",
                                        "1:2840:1.323",
                                        this);

            handleRollback(siTran);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            {
                SibTr.exception(tc, e);
                SibTr.exit(tc, "deliverExpressMessage", e);
            }

            throw e;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "deliverExpressMessage");
    }
506037
public void sendSilenceMessage(
     long startStamp,
     long endStamp,
     long completedPrefix,
     boolean requestedOnly,
     int priority,
     Reliability reliability,
     SIBUuid12 stream)
     throws SIResourceException
   {

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
       SibTr.entry(tc, "sendSilenceMessage",
         new Object[] {
           new Long(startStamp),
           new Long(endStamp),
           new Long(completedPrefix),
           new Integer(priority),
           reliability });

     ControlSilence sMsg;
     try
     {
       // Create new Silence message
       sMsg = _cmf.createNewControlSilence();
     }
     catch (Exception e)
     {
       // FFDC
       FFDCFilter.processException(
         e,
         "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendSilenceMessage",
         "1:787:1.164.1.5",
         this);

       SibTr.exception(tc, e);
       SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
         new Object[] {
           "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
           "1:794:1.164.1.5",
           e });

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

       throw new SIResourceException(
         nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.impl.PubSubOutputHandler",
            "1:805:1.164.1.5",
            e },
          null),
         e);
     }

     // As we are using the Guaranteed Header - set all the attributes as 
     // well as the ones we want.
     SIMPUtils.setGuaranteedDeliveryProperties(sMsg,
         _messageProcessor.getMessagingEngineUuid(), 
         null,
         stream,
         null,
         _destinationHandler.getUuid(),
         ProtocolType.PUBSUBINPUT,
         GDConfig.PROTOCOL_VERSION);
    
     sMsg.setStartTick(startStamp);
     sMsg.setEndTick(endStamp);
     sMsg.setPriority(priority);
     sMsg.setReliability(reliability);
     sMsg.setCompletedPrefix(completedPrefix);
     sMsg.setRequestedOnly(requestedOnly);
     
     // If the destination in a Link add Link specific properties to message
     if( _isLink )
     {
       sMsg = (ControlSilence)addLinkProps(sMsg);
     }
    
     // Send message to destination
     // Using MPIO

     // Send the message to the MessageTransmitter
     //add a target cellule to the array for sending
     SIBUuid8[] fromTo = new SIBUuid8[1];
     fromTo[0] = _targetMEUuid;
      
     // Send at priority+1 if this is a response to a Nack
     if( requestedOnly )
       _mpio.sendDownTree(fromTo, priority+1, sMsg);
     else
       _mpio.sendDownTree(fromTo, priority, sMsg);

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

   }
1267
public void sendAckMessage(
    SIBUuid8 meUuid,
    SIBUuid12 destUuid,
    SIBUuid8  busUuid,  
    long ackPrefix,
    int priority,
    Reliability reliability,
    SIBUuid12 streamID,
    boolean consolidate)
    throws SIResourceException
  {  
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendAckMessage",
        new Object[] { meUuid, 
                       new Long(ackPrefix),
                       new Integer(priority),
                       reliability,
                       streamID });

    ControlAck newAckMsg =
      createControlAckMessage(
        priority,
        reliability,
        streamID);
    
    newAckMsg.setAckPrefix(ackPrefix);        
    
    try
    {
      _parentInputHandler.handleControlMessage(null, newAckMsg);
    }
    catch (SIException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendAckMessage",
        "1:1731:1.164.1.5",
        this);
        
      SibTr.exception(tc, e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      {
        SibTr.exception(tc, e);
        SibTr.exit(tc, "sendAckMessage", e);
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendAckMessage");
  }
182275
@OnMessage
	public void onMessage(Dto msg, Session session) {
		try {
			messageBus.fire(session);
			messagePayloadBus.fire(msg);
		} catch (Exception e) {

			log.log(Level.SEVERE, e.getMessage(), e);
		}

	}
1266502
public void begin() throws HandlerException {
        if (_enabled) {
            try {
                _userTx = TransactionManagerLocator.INSTANCE.getUserTransaction();
                if (_userTx.getStatus() == Status.STATUS_NO_TRANSACTION) {
                    _userTx.begin();
                    _isInitiator = true;
                }
            } catch (SystemException se) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionBeginFailedSystem(se);
            } catch (NotSupportedException nse) {
                throw CommonKnowledgeMessages.MESSAGES.userTransactionBeginFailedNSE(nse);
            }
        }
    }
389389
@Override
  public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
          throws Exception {
    if (msg instanceof CreateStreamCommand) {
      createStream((CreateStreamCommand) msg, promise);
    } else if (msg instanceof SendGrpcFrameCommand) {
      sendGrpcFrame(ctx, (SendGrpcFrameCommand) msg, promise);
    } else if (msg instanceof CancelClientStreamCommand) {
      cancelStream(ctx, (CancelClientStreamCommand) msg, promise);
    } else if (msg instanceof SendPingCommand) {
      sendPingFrame(ctx, (SendPingCommand) msg, promise);
    } else if (msg instanceof GracefulCloseCommand) {
      gracefulClose(ctx, (GracefulCloseCommand) msg, promise);
    } else if (msg instanceof ForcefulCloseCommand) {
      forcefulClose(ctx, (ForcefulCloseCommand) msg, promise);
    } else if (msg == NOOP_MESSAGE) {
      ctx.write(Unpooled.EMPTY_BUFFER, promise);
    } else {
      throw new AssertionError("Write called for unexpected type: " + msg.getClass().getName());
    }
  }
1356223
private void localPut(
                          MessageItem msg,
                          TransactionCommon transaction)
                    throws
                    SIIncorrectCallException,
                    SIResourceException,
                    SINotPossibleInCurrentConfigurationException,
                    SINotAuthorizedException,
                    SILimitExceededException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "localPut",
                        new Object[] { msg, transaction });

        boolean stored = false;

        /*
         * If a non-empty frp exists at this point then send the message to the exception
         * destination
         */
        if (!msg.getMessage().isForwardRoutingPathEmpty())
        {
            // If we have a non-empty forward routing path then throw an exception. 
            // A topicspace can only be the final element in the path.

            SIMPIncorrectCallException e = new SIMPIncorrectCallException(
                            nls.getFormattedMessage(
                                                    "FORWARD_ROUTING_PATH_ERROR_CWSIP0249",
                                                    new Object[] { _destination.getName(),
                                                                  _messageProcessor.getMessagingEngineName() },
                                                    null));

            e.setExceptionReason(SIRCConstants.SIRC0037_INVALID_ROUTING_PATH_ERROR);
            e.setExceptionInserts(new String[] { _destination.getName(),
                                                _messageProcessor.getMessagingEngineName(),
                                                "unknown",
                                                SIMPUtils.getStackTrace(e) });

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            {
                SibTr.exception(tc, e);
                SibTr.exit(tc, "localPut", e);
            }

            throw e;

        }

        // If the message came in from a remote bus then
        // check whether the userId is authorised to access this destination
        if (msg.isFromRemoteBus())
        {
            // Check whether bus security is enabled 
            if (_messageProcessor.isBusSecure())
            {
                JsMessage jsMsg = msg.getMessage();

                // Before we test for anything else, see whether this message 
                // was sent by the privileged Jetstream SIBServerSubject. If it was
                // then we bypass the security checks
                if (!_messageProcessor.getAuthorisationUtils().sentBySIBServer(jsMsg))
                {
                    // Check authority to produce to destination
                    String userid = null;
                    if (_destination.isLink() && !_destination.isMQLink())
                    {
                        userid = ((LinkHandler) _destination).getInboundUserid();
                    }

                    // If the InboundUserid is null or we're not working with a link
                    // Set the userid from the message
                    if (userid == null)
                    {
                        // Use the id extracted from the message for access checks
                        userid = jsMsg.getSecurityUserid();
                    }

                    // Defect 240261: Map a null userid to an empty string
                    if (userid == null)
                    {
                        userid = ""; // Empty string means those users that have not
                                     // been authenticated, but who are still, of course,
                                     // members of EVERYONE. 
                    }

                    String discriminator = msg.getMessage().getDiscriminator();

                    // Create secContext from userid and discriminator
                    SecurityContext secContext = new SecurityContext(userid, discriminator);

                    if (!_destination.checkDestinationAccess(secContext,
                                                             OperationType.SEND))
                    {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "localPut", "not authorized to produce to this destination");

                        // Build the message for the Exception and the Notification
                        String nlsMessage =
                                        nls_cwsik.getFormattedMessage("DELIVERY_ERROR_SIRC_18", // USER_NOT_AUTH_SEND_ERROR_CWSIP0306
                                                                      new Object[] { _destination.getName(),
                                                                                    userid },
                                                                      null);

                        // Fire a Notification if Eventing is enabled
                        _messageProcessor.
                                        getAccessChecker().
                                        fireDestinationAccessNotAuthorizedEvent(_destination.getName(),
                                                                                userid,
                                                                                OperationType.SEND,
                                                                                nlsMessage);

                        // Thrown if user denied access to destination
                        SIMPNotAuthorizedException e = new SIMPNotAuthorizedException(nlsMessage);

                        e.setExceptionReason(SIRCConstants.SIRC0018_USER_NOT_AUTH_SEND_ERROR);
                        e.setExceptionInserts(new String[] { _destination.getName(),
                                                            userid });
                        throw e;
                    }

                    if (!_destination.
                                    checkDiscriminatorAccess(secContext,
                                                             OperationType.SEND))
                    {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "localPut", "not authorized to produce to this destination's discriminator");

                        // Write an audit record if access is denied
                        SibTr.audit(tc, nls_cwsik.getFormattedMessage(
                                                                      "DELIVERY_ERROR_SIRC_20", // USER_NOT_AUTH_SEND_ERROR_CWSIP0308
                                                                      new Object[] { _destination.getName(),
                                                                                    secContext.getDiscriminator(),
                                                                                    userid },
                                                                      null));

                        // Thrown if user denied access to destination 
                        SIMPNotAuthorizedException e = new SIMPNotAuthorizedException(
                                        nls_cwsik.getFormattedMessage(
                                                                      "DELIVERY_ERROR_SIRC_20", // USER_NOT_AUTH_SEND_ERROR_CWSIP0308
                                                                      new Object[] { _destination.getName(),
                                                                                    secContext.getDiscriminator(),
                                                                                    userid },
                                                                      null));

                        e.setExceptionReason(SIRCConstants.SIRC0020_USER_NOT_AUTH_SEND_ERROR);
                        e.setExceptionInserts(new String[] { _destination.getName(),
                                                            secContext.getDiscriminator(),
                                                            userid });
                        throw e;
                    }
                }
            }
        }

        boolean forcePut = msg.isForcePut();
        // Check SendAllowed for local case.
        //send allowed can be false ONLY if this msg is not force put
        boolean isSendAllowed = forcePut ||
                                (_itemStream.isSendAllowed() && _destination.isSendAllowed());

        if (!isSendAllowed)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(
                           tc,
                           "localPut",
                           "Destination send disallowed");

            SIMPNotPossibleInCurrentConfigurationException e = new SIMPNotPossibleInCurrentConfigurationException(
                            nls.getFormattedMessage("DESTINATION_SEND_DISALLOWED_CWSIP0253",
                                                    new Object[] { _destination.getName(),
                                                                  _messageProcessor.getMessagingEngineName() }, null));

            e.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
            e.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.PubSubInputHandler.localPut",
                                                "1:1334:1.329.1.1",
                                                SIMPUtils.getStackTrace(e) });
            throw e;
        }

        // check total items on itemstream not exceeded (allow extra 1 for ref stream)
        //but only if forcePut==false
        long topicSpaceHighLimit = _destination.getPublishPoint().getDestHighMsgs();
        if (!forcePut &&
            ((topicSpaceHighLimit != -1)
            && (_itemStream.getTotalMsgCount() >= topicSpaceHighLimit)
            ))
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(
                           tc,
                           "localPut",
                           "Destination reached high limit");

            SIMPLimitExceededException e = new SIMPLimitExceededException(
                            nls.getFormattedMessage("DESTINATION_HIGH_MESSAGES_ERROR_CWSIP0251",
                                                    new Object[] { _destination.getName(),
                                                                  new Long(topicSpaceHighLimit),
                                                                  _messageProcessor.getMessagingEngineName() },
                                                    null));
            e.setExceptionReason(SIRCConstants.SIRC0025_DESTINATION_HIGH_MESSAGES_ERROR);
            e.setExceptionInserts(new String[] { _destination.getName(),
                                                new Long(topicSpaceHighLimit).toString() });
            throw e;
        }

        if (msg.isTransacted())
        {
            // Dont store the msg till pre-prepare when we know if there are any
            // subscribers.  For now, just register for the pre-prepare callback on
            // the transaction.  //183715.1
            registerMessage(msg, transaction); //183715.1      
        }
        else
        {
            LocalTransaction siTran = _txManager.createLocalTransaction(false);

            // If COD reports are required, register for the precommit callback
            if (msg.getReportCOD() != null && _destination instanceof BaseDestinationHandler)
                msg.registerMessageEventListener(MessageEvents.COD_CALLBACK, (BaseDestinationHandler) _destination);

            try
            {
                // Perform the PubSub match.
                stored = localFanOut(msg, siTran, false);

                if (stored)
                    siTran.commit();
                else
                    siTran.rollback();
            } catch (RuntimeException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.PubSubInputHandler.localPut",
                                            "1:1397:1.329.1.1",
                                            this);

                SIMPErrorException ee = new SIMPErrorException(e);

                ee.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
                ee.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.ProducerSessionImpl.handleMessage",
                                                     "1:1404:1.329.1.1",
                                                     SIMPUtils.getStackTrace(e) });
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                {
                    SibTr.exception(tc, ee);
                    SibTr.exit(tc, "localPut", ee);
                }

                throw ee;

            } catch (SIRollbackException e)
            {
                // No FFDC code needed
                handleRollback(siTran);

                SIMPRollbackException ee = new SIMPRollbackException(e.getMessage());
                ee.setStackTrace(e.getStackTrace());
                ee.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
                ee.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.ProducerSessionImpl.handleMessage",
                                                     "1:1424:1.329.1.1",
                                                     SIMPUtils.getStackTrace(e) });

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "localPut", ee);

                throw ee;
            } catch (SIConnectionLostException e)
            {
                // No FFDC code needed
                handleRollback(siTran);

                SIMPConnectionLostException ee = new SIMPConnectionLostException(e.getMessage());
                ee.setStackTrace(e.getStackTrace());
                ee.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
                ee.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.ProducerSessionImpl.handleMessage",
                                                     "1:1441:1.329.1.1",
                                                     SIMPUtils.getStackTrace(e) });

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "localPut", ee);

                throw ee;
            } catch (SIIncorrectCallException e)
            {
                // No FFDC code needed
                handleRollback(siTran);

                SIMPIncorrectCallException ee = new SIMPIncorrectCallException(e.getMessage());
                ee.setStackTrace(e.getStackTrace());
                ee.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
                ee.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.ProducerSessionImpl.handleMessage",
                                                     "1:1458:1.329.1.1",
                                                     SIMPUtils.getStackTrace(e) });
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "localPut", ee);
                throw ee;
            } catch (SIResourceException e)
            {
                // No FFDC code needed
                handleRollback(siTran);

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

                SIMPResourceException ee = new SIMPResourceException(e);
                ee.setExceptionReason(SIRCConstants.SIRC0901_INTERNAL_MESSAGING_ERROR);
                ee.setExceptionInserts(new String[] { "com.ibm.ws.sib.processor.impl.ProducerSessionImpl.handleMessage",
                                                     "1:1475:1.329.1.1",
                                                     SIMPUtils.getStackTrace(e) });
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "localPut", ee);
                throw ee;
            }

            // Release the JsMessage from the parent MessageItem (Any MessageItemReferences will
            // have their own references to the JsMessage (unless they've released their's too)
            msg.releaseJsMessage();
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "localPut");
    }
854477
void sendToNeighbour(JsMessage message, Transaction transaction)
    throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendToNeighbour", new Object[] { message, transaction });

    if (iProducerSession == null)
      createProducerSession();

    try
    {
      iProducerSession.send((SIBusMessage) message, (SITransaction) transaction);

    }
    catch (SIException e)
    {
      // No FFDC code needed

      SibTr.exception(tc, e);
      SibTr.warning(tc, "PUBSUB_CONSISTENCY_ERROR_CWSIP0383", new Object[]{JsAdminUtils.getMENameByUuidForMessage(iMEUuid.toString()), e});


      SIResourceException ee = null;
      if (! (e instanceof SIResourceException))
        ee = new SIResourceException(e);
      else
        ee = (SIResourceException)e;


      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "sendToNeighbour", ee);

      throw ee;
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendToNeighbour");
  }
1013267
@Override
    public void sendMsg(String msg)
    {
        DbConn cnx = Helpers.getNewDbSession();
        try
        {
            Message.create(cnx, msg, ji.getId());
            cnx.commit();
        }
        finally
        {
            Helpers.closeQuietly(cnx);
        }
    }