Creates a new {@link TopicProducer}. The callback is called to convert an object that is sent into the TopicProducer
into a JMS message type.
972769
public <T> TopicProducer<T> createTopicProducer(final String topic, final ProducerCallback<T> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicProducer<T>(connectionFactory, jmsConfig, topic, messageCallback);
    }
326354
@Override
	public synchronized TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException
    {
        checkNotClosed();
        LocalTopicSession session =  new LocalTopicSession(idProvider.createID(),this,engine,transacted,acknowledgeMode);
        registerSession(session);
        return session;
    }
1213185
public MessagelistenerType<MessageadapterType<T>> getOrCreateMessagelistener()
   {
      List<Node> nodeList = childNode.get("messagelistener");
      if (nodeList != null &&  nodeList.size() > 0)
      {
         return new MessagelistenerTypeImpl<MessageadapterType<T>>(this, "messagelistener", childNode, nodeList.get(0));
      }
      return createMessagelistener();
   }
73301
@Override
    public ProducerSession createProducerSession(SIDestinationAddress destAddr,
                                                 DestinationType destinationType,
                                                 OrderingContext extendedMessageOrderingContext,
                                                 String alternateUser)

                    throws SIConnectionDroppedException,
                    SIConnectionUnavailableException,
                    SITemporaryDestinationNotFoundException,
                    SIErrorException,
                    SIIncorrectCallException,
                    SIResourceException,
                    SINotPossibleInCurrentConfigurationException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.entry(CoreSPIConnection.tc, "createProducerSession",
                        new Object[] { destAddr,
                                      destinationType,
                                      extendedMessageOrderingContext,
                                      alternateUser,
                                      this });

        // See if this connection has been closed
        checkNotClosed();

        checkProducerSessionNullParameters(destAddr);

        SecurityContext secContext = null;
        // If security is enabled then we need to set up a security context
        if (_isBusSecure)
        {
            // Feature 219101: Add the alternate user string to the security context
            secContext = new SecurityContext(_subject,
                            alternateUser,
                            null,
                            _messageProcessor.getAuthorisationUtils());
        }

        boolean keepSecurityUserid = false;

        ProducerSession producer =
                        internalCreateProducerSession(destAddr,
                                                      destinationType,
                                                      false,
                                                      secContext,
                                                      keepSecurityUserid,
                                                      false,
                                                      true,
                                                      true,
                                                      null);

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConnection.tc, "createProducerSession", producer);
        //Destination is on the local bus
        return producer;
    }
4505
public KafkaProducer<String, String> createStringProducer(Properties overrideConfig) {
        return createProducer(new StringSerializer(), new StringSerializer(), overrideConfig);
    }
638067
private Controllable createTopicMBean(Controllable c) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "createPublicationPointMBean", new Object[] { c });
        TopicImpl pp = new TopicImpl(_me, c);
        controllableMap.put(c, pp);
        try {
            Dictionary<String, Object> properties = new Hashtable<String, Object>();
            properties.put("service.vendor", "IBM");
            String cName = c.getName();
            properties.put("jmx.objectname", "WebSphere:feature=wasJmsServer,type=Topic,name=" + cName);
            ServiceRegistration<TopicMBean> topicMBean = (ServiceRegistration<TopicMBean>) bcontext.registerService(TopicMBean.class.getName(), pp, properties);
            serviceObjects.put(cName, topicMBean);
        } catch (Exception e) {
            SibTr.exception(tc, e);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "createPublicationPointMBean", new Object[] { pp });
        return pp;

    }
161388
@Override
    public JMSProducer createProducer() {

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

        JMSProducer jmsProducer = null;
        try {
            // Create messageproducer first, since we reuse most in jmsproducer
            // destination is passed as null because during send,destination is passed
            // jmssession.createproducer() is not used becuase it stores the list of producers
            // which is not required for simplified API's
            MessageProducer msgProducer = jmsSession.instantiateProducer(null);
            jmsProducer = new JmsJMSProducerImpl(msgProducer);
        } catch (JMSException e) {
            // should never have got here because we pass destination as null so there is no chance 
            // that instantiateProducer() will throw any exception

            FFDCFilter.processException(e,
                                        JmsJMSContextImpl.class.toString() + ".createProducer",
                                        "598", this);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createProducer");

        return jmsProducer;

    }
1123461
public <T> Producer<T> createProducer(final Bean<X> declaringBean, final Bean<T> bean, DisposalMethod<X, T> disposalMethod) {
        EnhancedAnnotatedMethod<T, X> enhancedMethod = getManager().getServices().get(MemberTransformer.class).loadEnhancedMember(method, getManager().getId());
        return new ProducerMethodProducer<X, T>(enhancedMethod, disposalMethod) {

            @Override
            public AnnotatedMethod<X> getAnnotated() {
                return method;
            }

            @Override
            public BeanManagerImpl getBeanManager() {
                return getManager();
            }

            @Override
            public Bean<X> getDeclaringBean() {
                return declaringBean;
            }

            @Override
            public Bean<T> getBean() {
                return bean;
            }
        };
    }
1090336
@Override
    public ProducerSession createProducerSession(
                                                 SIDestinationAddress destAddr,
                                                 String discriminator,
                                                 DestinationType destType,
                                                 OrderingContext extendedMessageOrderingContext,
                                                 String alternateUser)
                    throws SIConnectionDroppedException,
                    SIConnectionUnavailableException,
                    SITemporaryDestinationNotFoundException,
                    SIErrorException,
                    SIIncorrectCallException,
                    SIResourceException,
                    SINotPossibleInCurrentConfigurationException
    {
        ProducerSession session =
                        createProducerSession(destAddr,
                                              discriminator,
                                              destType,
                                              extendedMessageOrderingContext,
                                              alternateUser,
                                              false,
                                              false);

        return session;
    }
1101751
public BaseMessageHeader createMessageHeader()
    {
        Map<String,Object> properties = ((PropertiesField)this.getField(MessageLog.MESSAGE_HEADER_PROPERTIES)).getProperties();
        BaseMessageHeader messageHeader = null;
        String strMessageHeaderClassName = this.getField(MessageLog.MESSAGE_HEADER_CLASS_NAME).toString();
        String strQueueName = this.getField(MessageLog.MESSAGE_QUEUE_NAME).toString();
        String strQueueType = this.getField(MessageLog.MESSAGE_QUEUE_TYPE).toString();
        Object source = null;
        messageHeader = BaseMessageHeader.createMessageHeader(strMessageHeaderClassName, strQueueName, strQueueType, source, properties);
        if (messageHeader == null)
            messageHeader = new TrxMessageHeader(null, properties);
        if (messageHeader instanceof TrxMessageHeader)
        {
            properties = ((PropertiesField)this.getField(MessageLog.MESSAGE_INFO_PROPERTIES)).getProperties();
            ((TrxMessageHeader)messageHeader).setMessageInfoMap(properties);
            properties = ((PropertiesField)this.getField(MessageLog.MESSAGE_TRANSPORT_PROPERTIES)).getProperties();
            ((TrxMessageHeader)messageHeader).setMessageTransportMap(properties);
            ((TrxMessageHeader)messageHeader).put(TrxMessageHeader.LOG_TRX_ID, this.getCounterField().toString());  // Should be there
        }
        return messageHeader;
    }
587465
@Override
	public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        
	        RemoteTopicSubscriber subscriber =  new RemoteTopicSubscriber(idProvider.createID(),
	        		                                                      this,
	                                                                      DestinationTools.asRef(topic),
	                                                                      messageSelector,
	                                                                      noLocal);
	        registerConsumer(subscriber);
	        subscriber.remoteInit();
	        return subscriber;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
297717
public final SubscriptionMessage createNewSubscriptionMessage() throws MessageCreateFailedException {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "createNewSubscriptionMessage");
    SubscriptionMessage msg = null;
    try {
      msg = new SubscriptionMessageImpl(MfpConstants.CONSTRUCTOR_NO_OP);
    }
    catch (MessageDecodeFailedException e) {
      /* No need to FFDC this as JsMsgObject will already have done so */
      // No FFDC code needed
      throw new MessageCreateFailedException(e);
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "createNewSubscriptionMessage");
    return msg;
  }
72426
@Override
    public TopicConnectionFactory createTopicConnectionFactory(JmsJcaConnectionFactory jcaConnectionFactory) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "createTopicConnectionFactory", jcaConnectionFactory);
        TopicConnectionFactory tcf = new JmsManagedTopicConnectionFactoryImpl(jcaConnectionFactory);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "createTopicConnectionFactory", tcf);
        return tcf;
    }
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);

    }
242471
@Override
    public MapMessage createMapMessage() throws JMSRuntimeException {

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

        MapMessage mapMessage = null;

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

        return mapMessage;

    }
449864
@Override
    public TopicSubscriber createDurableSubscriber(Topic topic, String name) throws JMSException {
        throw new JMSException(SQSMessagingClientConstants.UNSUPPORTED_METHOD);
    }
1312529
public <K, V> void produce(String topic, KafkaProducer<K, V> producer, Map<K, V> data) {
        data.forEach((k, v) -> producer.send(new ProducerRecord<>(topic, k, v)));
        producer.flush();
    }