Creates a new {@link TopicProducer}. The producer accepts strings and sends them as a text message.
769668
public TopicProducer<String> createTopicTextMessageProducer(final String topic)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicProducer<String>(connectionFactory, jmsConfig, topic, new TextMessageProducerCallback());
    }
247210
protected final void registerProducer( AbstractMessageProducer producer )
    {
        if (producersMap.put(producer.getId(),producer) != null)
        	throw new IllegalArgumentException("Producer "+producer.getId()+" already exists");
    }
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();
    	}
    }
695091
@Override
	public MessageProducer createProducer(Destination destination) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalMessageProducer producer = new LocalMessageProducer(this,destination,idProvider.createID());
	        registerProducer(producer);
	        return producer;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
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();
    }
587076
@Override
	public TopicSubscriber createSubscriber(Topic topic, String messageSelector, boolean noLocal) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalTopicSubscriber subscriber = new LocalTopicSubscriber(engine,this,topic,messageSelector,noLocal,idProvider.createID(),null);
	        registerConsumer(subscriber);
	        subscriber.initDestination();
	        
	        return subscriber;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
1265795
protected ProducerCloseable<String, byte[]> createProducer(ProducerConfig config) {
    return this.closer.register(new ProducerCloseable<String, byte[]>(config));
  }
703971
public static Message create( String text, Object data, ProcessingUnit owner  )
    {
        return new SimpleMessage( text, data, owner);
    }
724951
@Override
	public final void publish(Topic topic, Message message) throws JMSException
    {
        send(topic,message);
    }
583917
public boolean send(String topic, String message) throws IOException {
        List<Session> sessions = topicSubscribers.get(topic);
        if (sessions != null) {
            for (Session session : sessions) {
                session.getBasicRemote().sendText(message);
            }
        }
        return sessions != null;
    }
1316723
private static final String createMessage(Class<? extends IckleActivity> injectorActivity,
											  Class<? extends Annotation> annotation,
											  Class<? extends Object> expectedType,
											  Class<? extends Object> valueType,
											  Field field) {
		
		StringBuilder stringBuilder = new StringBuilder();
		
		stringBuilder.append("Annotation ");
		stringBuilder.append(annotation.getName() );
		stringBuilder.append(" is illegally used on field ");
		stringBuilder.append(field.getName());
		stringBuilder.append(" of type ");
		stringBuilder.append(valueType.getName());
		stringBuilder.append(" in ");
		stringBuilder.append(injectorActivity.getName());
		stringBuilder.append(". The expected field type is ");
		stringBuilder.append(expectedType.getName());
		stringBuilder.append( ". ");
	
		return stringBuilder.toString();
	}
47783
public PubsubFuture<List<String>> publish(final String project, final String topic,
                                            final List<Message> messages) {
    return publish0(messages, Topic.canonicalTopic(project, topic));
  }
495660
private void sendUsingProducerSession(Message message) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "sendUsingProducerSession", new Object[] { message });

        checkClosed();
        // check for sync/async conflicts
        session.checkSynchronousUsage("send");
        // if the supplied message is null, throw a jms MessageFormatException.
        if (message == null) {
            throw (MessageFormatException) JmsErrorUtils.newThrowable(
                                                                      MessageFormatException.class,
                                                                      "INVALID_VALUE_CWSIA0068",
                                                                      new Object[] { "message", null },
                                                                      tc);
        }
        if (propsOverriden) {
            // This block will be invoked if the user is flipping back and forth
            // between the multi-arg identified send, and this 1-arg send. In this
            // case we need to re-instate the dm, pri, ttl values from the producer.
            producerProperties.setInDeliveryMode(defaultDeliveryMode);
            producerProperties.setInPriority(defaultPriority);
            producerProperties.setInTTL(defaultTimeToLive);

            // Mark that we are back with the defaults so we don't do this set the
            // next time around.
            propsOverriden = false;
        }

        // Delegate to the internal send method.
        sendMessage(producerProperties, message, dest);

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

    }
8
public TopicProducer<Object> createTopicJsonProducer(final String topic)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicProducer<Object>(connectionFactory, jmsConfig, topic, producerCallback);
    }
1306832
public MessageId send(ProducerConnectionContext context, BasicMessage basicMessage) throws JMSException {
        return send(context, basicMessage, null);
    }
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));
    }
36713
@Override
    public Producer createProducer() throws Exception {
        LOG.debug(this.toString());

        if (direction == null || (direction.compareTo("") == 0)) {
            direction = "DIR_OUTPUT_ONLY";
        }
        if (mode == null || (mode.compareTo("") == 0)) {
            mode = "MODE_OUTPUT_PUSH_PULL";
        }
        if (trigger == null || (trigger.compareTo("") == 0)) {
            trigger = "TRIGGER_NONE";
        }

        int internalDirection = internalValueWithOR(direction);
        int internalMode = internalValueWithOR(mode);
        int internalTrigger = internalValueWithOR(trigger);

        GPIOPinConfig pinConfig = new GPIOPinConfig(DeviceConfig.DEFAULT, Integer.parseInt(this.gpioId),
                internalDirection, internalMode, internalTrigger, state);
        GPIOPin pin = DeviceManager.open(GPIOPin.class, pinConfig);

        return new GPIOProducer(this, pin);
    }