code: @SuppressWarnings("rawtypes")
  	public void open(Map conf, TopologyContext context,
  			SpoutOutputCollector collector) {
  		if(this.jmsProvider == null){
  			throw new IllegalStateException("JMS provider has not been set.");
  		}
  		if(this.tupleProducer == null){
  			throw new IllegalStateException("JMS Tuple Producer has not been set.");
  		}
  		Integer topologyTimeout = (Integer)conf.get("topology.message.timeout.secs");
  		// TODO fine a way to get the default timeout from storm, so we're not hard-coding to 30 seconds (it could change)
  		topologyTimeout = topologyTimeout == null ? 30 : topologyTimeout;
  		if( (topologyTimeout.intValue() * 1000 )> this.recoveryPeriod){
  		    LOG.warn("*** WARNING *** : " +
  		    		"Recovery period ("+ this.recoveryPeriod + " ms.) is less then the configured " +
  		    		"'topology.message.timeout.secs' of " + topologyTimeout + 
  		    		" secs. This could lead to a message replay flood!");
  		}
  		this.queue = new LinkedBlockingQueue<Message>();
  		this.toCommit = new TreeSet<JmsMessageID>();
          this.pendingMessages = new HashMap<JmsMessageID, Message>();
  		this.collector = collector;
  		try {
  			ConnectionFactory cf = this.jmsProvider.connectionFactory();
  			Destination dest = this.jmsProvider.destination();
  			this.connection = cf.createConnection();
  			this.session = connection.createSession(false,
  					this.jmsAcknowledgeMode);
  			MessageConsumer consumer = session.createConsumer(dest);
  			consumer.setMessageListener(this);
  			this.connection.start();
  			if (this.isDurableSubscription() && this.recoveryPeriod > 0){
  			    this.recoveryTimer = new Timer();
  			    this.recoveryTimer.scheduleAtFixedRate(new RecoveryTask(), 10, this.recoveryPeriod);
  			}
  			
  		} catch (Exception e) {
  			LOG.warn("Error creating JMS connection.", e);
  		}
  
  	}
docs: <code>ISpout</code> implementation.
  <p/>
  Connects the JMS spout to the configured JMS destination
  topic/queue.