Lexical Feature Stream

 curl --data "api_key=***bf3957e823316e5dc18c031c6ddc8074b***&function=stream" "https://api3.socialmarketanalytics.com/api/detail?subject=all&ontology=ticker&items=feature&sourcestring=twitter&dates=datetime+eq+realtime&timezone=UTC" 
                    

     
package sma.stream;
 
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sma.stream.facility.URIFactory;
import sma.stream.facility.URIFactoryImpl;
import sma.stream.facility.URLConnectionFactory;
import sma.stream.facility.URLConnectionFactoryImpl;
import sma.stream.model.*;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.*;
 
/**
 * Copyright (c) 2015 Social Market Analytics 
 *
 * All rights reserved.
 */
 
 
/**
 * Application to parse json response from SMA stream.
 *
 * 

 * To Run Main
 * 
 * In order to run application:
 * 

 * The following command line arguments are required.
 * 
 API Key Param: 40 characters long string
 * 
 API Function Param: e.g. stream
 * 
 API Subject Param: e.g. all
 * 
 API Ontology Param: e.g. ticker
 * 
 API Items Param: e.g. sentiment
 * 

 * @author Hisham Javed
 *
 */
 
public class SMASentimentStreamConsumer {
    /** Logger */
    protected static Logger logger = LoggerFactory.getLogger(SMASentimentStreamConsumer.class);
 
    // SMA stream feed
    /** {@link sma.stream.facility.URLConnectionFactory} to manage interactions with SMA Stream */
    URLConnectionFactory urlFactory;
 
    /**
     * Main Constructor
     * @param urlFactory
     * @throws java.io.IOException
     */
    public SMASentimentStreamConsumer(URLConnectionFactory urlFactory) throws IOException {
        this.urlFactory = urlFactory;
    }
 
    public URLConnectionFactory getURLConnectionFactory() { return urlFactory; }
 
    public void run() {
        try {
            System.out.println("Begin stream");
            InputStream is = this.urlFactory.createStreamConnection();
            BufferedReader in = new BufferedReader(new InputStreamReader(is));
            String line = null;
            /**
             * JSON parser.
             */
            JSONParser parser = new JSONParser();
            while((line = in.readLine()) != null) {
                if(!line.equals("")){
                    try{
                        System.out.println(line);
                        JSONObject jsonObject =(JSONObject) parser.parse(line);
                        Sentiment sentiment = new Sentiment((JSONObject)jsonObject.get("SENTIMENT"));
                        if(sentiment!=null){
                            System.out.println("Parsed successfully");
                        }
                    }
                    catch(ParseException ex){
                        logger.error("Invalid json object");
                    }
                }
            }
            System.out.println("End stream");
 
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("End Stream with exception");
            System.exit(1);
        }
    }
 
 
    /**
     * @param args
     */
    public static void main(String[] args) {
        Date startDate = new Date();
        int exitValue = 0;
        if (args.length < 5) {
            String className = Thread.currentThread().getStackTrace()[1].getClassName();
            System.err.println("Usage: " + className + " domain topic systemName apiKey function subject ontology");
            System.exit(2);
        }
 
        try
        {
            //SMA Sample Arguments
            //"5723c1397f109dd7d62515b49b8b74e9a263c386" "stream" "all" "ticker" "feature"
 
            //Use the following VM Param in case of SSLProtocolException to connect with SMA API only in development env.
            //-Djsse.enableSNIExtension=false
 
 
            // Stream params
            String apiKey = args[0]; // apiKey:40 characters long string
            String function = args[1]; // function:stream
            String subject = args[2]; // subject:all
            String ontology = args[3]; // ontology:ticker
            String queryOptions = args[4]; //items:feature
 
            // Create SMA API URL Connection Factory
            URIFactory uriFactory = new URIFactoryImpl(subject,ontology,queryOptions);
            URLConnectionFactory urlFactory = new URLConnectionFactoryImpl(uriFactory,apiKey,function);
 
            // Create Extractor
            SMASentimentStreamConsumer consumer = new SMASentimentStreamConsumer(urlFactory);
            consumer.run();
        }
        catch(Exception ex)
        {
            exitValue = 1;
            ex.printStackTrace(System.err);
        }
        finally
        {
            Date endDate = new Date();
            System.out.println("Exiting after " + (endDate.getTime() - startDate.getTime()) / 1000 + " seconds ...");
            System.exit(exitValue);
        }
    }
}
/* URIFactoryImpl.java */
        package sma.stream.facility;


        import java.net.MalformedURLException;
        import java.net.URI;

        /**
        * Implementation of {@link URIFactory} that creates {@link URI}s to connect
        * to the SMA end-points.
        */

        /**
        * @author Hisham Javed
        *
        */
        public class URIFactoryImpl implements URIFactory {

        public static final String BASE_SMA_STREAM_URI = "https://api3.socialmarketanalytics.com/api/detail?subject=%s&ontology=%s&items=%s";

        final String subject;
        final String ontology;
        final String queryOptions;

        /**
        * Default constructor.
        */
        public URIFactoryImpl()
        {
        this("all", "ticker","feature");
        }

        /**
        *
        * @param subject
        * @param ontology
        * @param queryOptions
        */
        public URIFactoryImpl(String subject, String ontology,String queryOptions)
        {
        this.subject = subject.trim();
        this.ontology = ontology.trim();
        this.queryOptions = queryOptions.trim();
        }

        @Override
        public URI createStreamURI() {
        return createSourceStreamURI(subject, ontology, queryOptions);
        }

        public URI createSourceStreamURI(String subject, String ontology,String queryOptions) {
        if (subject == null || subject.trim().isEmpty()) {
        throw new IllegalArgumentException("The subject cannot be null or empty");
        }
        if (ontology == null || ontology.trim().isEmpty()) {
        throw new IllegalArgumentException("The ontology cannot be null or empty");
        }
        if (queryOptions == null || queryOptions.trim().isEmpty()) {
        throw new IllegalArgumentException("The queryOptions cannot be null or empty");
        }

        return URI.create(String.format(BASE_SMA_STREAM_URI, subject.trim(), ontology.trim(),queryOptions.trim()));
        }


        /**
        * Unit test
        * @param args
        */
        public static void main(String[] args) {
        URIFactory uriFactory = new URIFactoryImpl();
        try {
        URI uri = uriFactory.createStreamURI();
        System.out.println("Default Stream URL: " + uri.toURL().toString());

        } catch (MalformedURLException e) {
        e.printStackTrace();
        System.exit(1);
        }
        System.exit(0);
        }
        }

/* URLConnectionFactoryImpl.java */
        package sma.stream.facility;

        import org.slf4j.Logger;
        import org.slf4j.LoggerFactory;
        import javax.naming.AuthenticationException;
        import java.io.*;
        import java.net.*;
        import java.util.zip.GZIPInputStream;
        import java.util.zip.Inflater;
        import java.util.zip.InflaterInputStream;

        /**
        * A factory to create {@link java.net.URI}s to connect to SMA API Stream.
        *
        * @author Hisham Javed
        *
        */
        public class URLConnectionFactoryImpl implements URLConnectionFactory {

        private final Logger logger = LoggerFactory.getLogger(getClass());
        final String USER_AGENT = "SMA (https://www.socialmarketanalytics.com/)";
        final int connectTimeout = 10000;
        final int readTimeout = 30000; // Recommended a 30 second read timeout

        /**
        * Default query URL.
        */
        protected URIFactory uriFactory;

        protected String apiKey;

        protected String function = "all";

        /**
        * Standard HTTP query options.
        */
        protected String queryOptions = null;

        /**
        * Instantiate default {@link tm.facility.stream.URLConnectionFactoryImpl} with default connection parameters.
        */
        public URLConnectionFactoryImpl() {
        }

        /**
        * Instantiate {@link tm.facility.stream.URLConnectionFactoryImpl} with customized OAuth connection parameters.
        */
        public URLConnectionFactoryImpl(URIFactory factory,String apiKey)
        {
        this.uriFactory = factory;
        this.apiKey = apiKey;
        }

        /**
        * Instantiate {@link tm.facility.stream.URLConnectionFactoryImpl} with customized OAuth connection parameters.
        */
        public URLConnectionFactoryImpl(URIFactory factory,String apiKey,String function)
        {
        this.uriFactory = factory;
        this.apiKey=apiKey;
        this.function=function;
        }


        /** validate responses
        * @throws java.io.IOException
        * @throws javax.naming.AuthenticationException */
        public final void validateStatusLine(final URI uri, final int statusCode, final String reason) throws IOException, AuthenticationException {
        logger.info(uri.toURL().toString());
        if (statusCode == 200 || statusCode == 201) {
        // nothing to do
        } else if (statusCode == 401) {
        String message = String.format("Connection to %s: AuthenticationException code: %s %s",
        uri, statusCode, reason);
        logger.info(message);
        throw new AuthenticationException(message);
        } else {
        String message = String.format("Connection to %s: Unexpected status code: %s %s",
        uri, statusCode, reason);
        logger.info(message);
        throw new IOException(message);
        }
        }
        protected void doConfiguration(final URLConnection uc) {
        // Not sure what to do here.
        }
        public final InputStream getResource(final URI uri) throws MalformedURLException, IOException {
        HttpURLConnection huc = (HttpURLConnection) uri.toURL().openConnection();
        huc.setAllowUserInteraction(false);
        huc.setDefaultUseCaches(false);
        huc.setConnectTimeout(connectTimeout);
        huc.setReadTimeout(readTimeout);
        huc.setRequestMethod("POST");
        huc.setRequestProperty("Content-Type", "application/x-www-form-urlencoded");
        huc.setRequestProperty("Accept-Encoding", "gzip,deflate");
        huc.setRequestProperty("User-Agent", USER_AGENT);
        huc.setRequestProperty("Content-Language", "en-US");
        String urlParameters = "api_key="+ URLEncoder.encode(apiKey,"UTF-8") +"&function="+URLEncoder.encode(function,"UTF-8");
        huc.setRequestProperty("Content-Length", "" + Integer.toString(urlParameters.getBytes().length));
        huc.setUseCaches(false);
        huc.setDoOutput(true);
        huc.setDoInput(true);

        DataOutputStream wr = new DataOutputStream (
        huc.getOutputStream ());
        wr.writeBytes(urlParameters);
        wr.flush ();
        wr.close ();

        doConfiguration(huc);

        huc.connect();

        if(huc != null) {
        try {
        validateStatusLine(uri, huc.getResponseCode(), huc.getResponseMessage());
        } catch (AuthenticationException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        logger.info(e.getExplanation());
        throw new IOException(e.getMessage());
        };
        }
        InputStream is = huc.getInputStream();
        final String encoding = huc.getContentEncoding();
        if (encoding != null && encoding.equalsIgnoreCase("gzip")) {
        is = new GZIPInputStream(is);
        } else if (encoding != null && encoding.equalsIgnoreCase("deflate")) {
        is = new InflaterInputStream(is, new Inflater(true));
        }
        return is;
        }


        @Override
        public InputStream createStreamConnection() throws MalformedURLException, IOException {
        URI uri = uriFactory.createStreamURI();

        return getResource(uri);
        }
        }