title: Storm > Howto Integrate Java callback into a Spout
body: <p>I'm trying to integrate Storm (<a href="https://github.com/nathanmarz/storm/wiki/Clojure-DSL" rel="nofollow noreferrer">see here</a>) into my project. I grok the concepts of topologies, spouts, and bolts. But now, I'm trying to figure out the actual implementation of a few things. </p>
  
  <p><strong>A)</strong> I have a polyglot environment with Java and Clojure. My Java code is a callback class with methods firing streaming data. The event data pushed to those methods, is what I want to use as a spout. </p>
  
  <p>So the first question is how to connect the data coming into those methods, to a spout ? I'm trying to <strong>i)</strong> pass an <em>backtype.storm.topology.IRichSpout</em> , then <strong>ii)</strong> pass a <em>backtype.storm.spout.SpoutOutputCollector</em> (<a href="https://github.com/nathanmarz/storm/blob/master/src/jvm/backtype/storm/spout/SpoutOutputCollector.java" rel="nofollow noreferrer">see here</a>) to that spout's <strong><em>open</em></strong> function (<a href="https://github.com/nathanmarz/storm/blob/master/src/jvm/backtype/storm/spout/ISpout.java#L40" rel="nofollow noreferrer">see here</a>). But I can't see a way to actually pass in any kind of map or list. </p>
  
  <p><strong>B)</strong> The rest of my project is all Clojure. There will be a lot of data coming through those methods. Each event will have an ID of between 1 and 100. In Clojure, I'll want to split data coming from the spout, into different threads of execution. Those, I think, will be the bolts. </p>
  
  <p>How can I set up a Clojure bolt to take event data from the spout, then break-off a thread based on the ID of the incoming event ?</p>
  
  <p>Thanks in advance
  Tim</p>
  
  <p><strong>[EDIT 1]</strong></p>
  
  <p>I've actually gotten past this problem. I ended up <strong>1)</strong> implementing my own IRichSpout. I then <strong>2)</strong> connected that spout's internal tuple to the incoming stream data in my java callback class. I'm not sure if this is idiomatic. But it compiles and runs without error. However, <strong>3)</strong> I don't see the incoming stream data (definitely there), coming through the <strong><em>printstuff</em></strong> bolt. </p>
  
  <p>In order to ensure that the event data gets propagated, is there something specific I have to do in the spout or bolt implementation or topology definition? Thanks.</p>
  
  <pre>
  
        ;; tie Java callbacks to a Spout that I created
        (.setSpout java-callback ibspout)
  
        (storm/defbolt printstuff ["word"] [tuple collector]
          (println (str "printstuff --> tuple["tuple"] > collector["collector"]"))
        )
        (storm/topology
         { "1" (storm/spout-spec ibspout)
         }
         { "3" (storm/bolt-spec  { "1" :shuffle }
                                 printstuff
               )
         })
  
  </pre>
  
  <p><strong>[EDIT 2]</strong></p>
  
  <p>On the advice of SO member <a href="https://stackoverflow.com/users/553209/ankur">Ankur</a>, I'm rejigging my topology. After I've created my Java callback, I pass it's tuple to the below IBSpout, using <code>(.setTuple ibspout (.getTuple java-callback))</code>. I don't pass the entire Java callback object, because I get a NotSerializable error. Everything compiles and runs without error. But again, there's no data coming to my <strong><em>printstuff</em></strong> bolt. Hmmm.</p>
  
  <pre>
  
      public class IBSpout implements IRichSpout {
  
        /**
         * Storm spout stuff
         */
        private SpoutOutputCollector _collector;
  
        private List _tuple = new ArrayList();
        public void setTuple(List tuple) { _tuple = tuple; }
        public List getTuple() { return _tuple; }
  
        /**
         * Storm ISpout interface functions
         */
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
          _collector = collector;
        }
        public void close() {}
        public void activate() {}
        public void deactivate() {}
        public void nextTuple() {
          _collector.emit(_tuple);
        }
        public void ack(Object msgId) {}
        public void fail(Object msgId) {}
  
  
        public void declareOutputFields(OutputFieldsDeclarer declarer) {}
        public java.util.Map  getComponentConfiguration() { return new HashMap(); }
  
      }
  
  </pre>