Spouts - Getting Started with Storm - page 32

unknown device is one that is added after the topology is already running. A well-
known device group is one in which all devices in the group are known at start time.
As an example, create a spout to read the Twitter stream using the Twitter streaming
API . The spout will connect directly to the API, which serves as the message emitter.
Filter the stream to get all public tweets that match the track parameter (as documented
on the Twitter dev page). The complete example can be found at Twitter Example
github page.
The spout gets the connection parameters from the configuration object ( track , user ,
and password ) and creates a connection to the API (in this case, using the DefaultHttp-
Client from Apache ). It reads the connection one line at a time, parses the line from
JSON format into a Java object, and emits it.
public void nextTuple () {
//Create the client call
client = new DefaultHttpClient ();
client . setCredentialsProvider ( credentialProvider );
HttpGet get = new HttpGet ( STREAMING_API_URL + track );
HttpResponse response ;
try {
//Execute
response = client . execute ( get );
StatusLine status = response . getStatusLine ();
if ( status . getStatusCode () == 200 ){
InputStream inputStream = response . getEntity (). getContent ();
BufferedReader reader = new BufferedReader ( new InputStreamReader ( inputStream ));
String in ;
//Read line by line
while (( in = reader . readLine ())!= null ){
try {
//Parse and emit
Object json = jsonParser . parse ( in );
collector . emit ( new Values ( track , json ));
} catch ( ParseException e ) {
LOG . error ( "Error parsing message from twitter" , e );
}
}
}
} catch ( IOException e ) {
LOG . error ( "Error in communication with twitter api [" + get . getURI (). toString ()+ "],
sleeping 10s" );
try {
Thread . sleep ( 10000 );
} catch ( InterruptedException e1 ) {
}
}
}