public MessageProducer getOrCreateProducer ( final String topic ) { if ( ! this . shareProducer ) { FutureTask < MessageProducer > task = this . producers . get ( topic ) ; if ( task == null ) { task = new FutureTask < MessageProducer > ( new Callable < MessageProducer > ( ) { @ Override public MessageProducer call ( ) throws Exception { MessageProducer producer = MetaqTemplate . this . messageSessionFactory . createProducer ( ) ; producer . publish ( topic ) ; if ( ! StringUtils . isBlank ( MetaqTemplate . this . defaultTopic ) ) { producer . setDefaultTopic ( MetaqTemplate . this . defaultTopic ) ; } return producer ; } } ) ; FutureTask < MessageProducer > oldTask = this . producers . putIfAbsent ( topic , task ) ; if ( oldTask != null ) { task = oldTask ; } else { task . run ( ) ; } } try { MessageProducer producer = task . get ( ) ; return producer ; } catch ( ExecutionException e ) { throw ThreadUtils . launderThrowable ( e . getCause ( ) ) ; } catch ( InterruptedException e ) { Thread . currentThread ( ) . interrupt ( ) ; } } else { if ( this . sharedProducer == null ) { synchronized ( this ) { if ( this . sharedProducer == null ) { this . sharedProducer = this . messageSessionFactory . createProducer ( ) ; if ( ! StringUtils . isBlank ( this . defaultTopic ) ) { this . sharedProducer . setDefaultTopic ( this . defaultTopic ) ; } } } } this . sharedProducer . publish ( topic ) ; return this . sharedProducer ; } throw new IllegalStateException ( "Could not create producer for topic '" + topic + "'" ) ; }
Returns or create a message producer for topic .