code: @Override
  	public void prepare(Map stormConf, TopologyContext context,
  			OutputCollector collector) {
  		if(this.jmsProvider == null || this.producer == null){
  			throw new IllegalStateException("JMS Provider and MessageProducer not set.");
  		}
  		this.collector = collector;
  		LOG.debug("Connecting JMS..");
  		try {
  			ConnectionFactory cf = this.jmsProvider.connectionFactory();
  			Destination dest = this.jmsProvider.destination();
  			this.connection = cf.createConnection();
  			this.session = connection.createSession(this.jmsTransactional,
  					this.jmsAcknowledgeMode);
  			this.messageProducer = session.createProducer(dest);
  			
  			connection.start();
  		} catch (Exception e) {
  			LOG.warn("Error creating JMS connection.", e);
  		}	
  	}
docs: Initializes JMS resources.