CometClient.java

package essai;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Hashtable;
//import java.util.regex.Matcher;
//import java.util.regex.Pattern;

import org.apache.http.HttpResponse;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.conn.ClientConnectionManager;
import org.apache.http.conn.scheme.PlainSocketFactory;
import org.apache.http.conn.scheme.Scheme;
import org.apache.http.conn.scheme.SchemeRegistry;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.impl.conn.tsccm.ThreadSafeClientConnManager;
import org.apache.http.params.BasicHttpParams;
import org.apache.http.params.HttpParams;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

public class CometClient {

    private int             requestId;
    private String              cometUri;
    private String              clientId;
    private HttpClient          httpClient;
    private HttpPost            httpPost;
    private HttpParams          httpParams;
    private HttpResponse            httpResponse;
    private SchemeRegistry          schemeRegistry;
    private ClientConnectionManager     clientConnectionManager;
    private InputStreamReader       inputStreamReader;
    private InputStreamHandler      inputStreamHandler;

    public  enum                ConnectionStatus {
        DISCONNECTED, HANDSHAKE_ERROR, HANDSHAKE_OK, CONNECTING, CONNECTION_ERROR, CONNECTED
    };

    public CometClient(String uri) {
        this.cometUri   = uri + "/cometd";
        this.requestId  = 1;

        this.httpParams = new BasicHttpParams();
        this.schemeRegistry = new SchemeRegistry();
        this.schemeRegistry.register(new Scheme("http", 80, PlainSocketFactory.getSocketFactory()));
        this.clientConnectionManager = new ThreadSafeClientConnManager( this.schemeRegistry);
        this.httpClient = new DefaultHttpClient(this.clientConnectionManager, this.httpParams);
        this.httpPost   = new HttpPost(cometUri);
        this.httpPost.setHeader("Content-Type","application/json; charset=UTF-8");

    }

    public String getClientId() {
        return this.clientId;
    }

    public int getRequestId() {
        return this.requestId;
    }

    public InputStreamReader getInputStreamReader() {
        return this.inputStreamReader;
    }

    public boolean handshake() {

        JSONArray       jsonRequest;
        JSONArray       jsonResponse;
        JSONObject      jsonHandshake;

        String          request;
        String          result;
        BufferedReader  buffer;

        // Build JSON handshake request
        System.out.print("handshake(): building JSON handshake request... ");
        try {

            ArrayList<String> jsonSCT = new ArrayList<String>(1);
            jsonSCT.add("streaming");

            Hashtable<String, String> jsonEXT = new Hashtable<String,String>();

            // Replace with valid values before use:
            jsonEXT.put("mac","xx:xx:xx:xx:xx:xx");
            jsonEXT.put("rev","x");
            jsonEXT.put("uuid","xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx");

            jsonHandshake = new JSONObject()
                .put("channel","/meta/handshake")
                .put("ext", jsonEXT)
                .put("supportedConnectionTypes", jsonSCT)
                .put("version","1.0");

            jsonRequest = new JSONArray()
                .put(jsonHandshake);

            request = jsonRequest.toString();
            System.out.println("done.\n\t" + request);

        } catch (JSONException e) {

            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);
        }

        // Post JSON handshake request
        System.out.print("handshake(): posting request... ");
        try {

            this.httpPost.setEntity(new StringEntity(request,"UTF-8"));
            this.httpResponse = this.httpClient.execute(this.httpPost);
            System.out.println("done.");

        } catch(UnsupportedEncodingException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(ClientProtocolException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IOException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        }

        // Get Cometd server's response
        System.out.print("handshake(): awaiting response... ");
        try {
            buffer = new BufferedReader(new InputStreamReader(this.httpResponse.getEntity().getContent(),"UTF-8"));
            result = buffer.readLine();
            System.out.println("done.\n\t" + result);
        } catch(UnsupportedEncodingException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IOException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        }

        // Parse response
        System.out.print("handshake(): parsing response...");
        try {
            jsonResponse = new JSONArray(result);
            jsonHandshake = jsonResponse.getJSONObject(0);
            System.out.println("done");
        } catch ( JSONException e ) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);
        }

        // Check response
        System.out.println("handshake(): checking response:");
        try {
            System.out.print("- channel: ");
            System.out.print(jsonHandshake.getString("channel") + " ");
            if( jsonHandshake.getString("channel").equals("/meta/handshake") ) {
                System.out.println("OK");
            } else {
                System.out.println("ERROR: Unexpected channel!");
                return(false);
            }

            System.out.print("- successful: ");
            System.out.print(jsonHandshake.getBoolean("successful") + " ");
            if( jsonHandshake.getBoolean("successful") == true ) {
                System.out.println("OK");
            } else {
                System.out.println("ERROR: Unexpected result!");
                return(false);
            }

            System.out.print("- clientId: ");
            System.out.print(jsonHandshake.getString("clientId") + " ");
            this.clientId = jsonHandshake.getString("clientId");
            System.out.println("OK");

        } catch ( JSONException e ) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);
        }

        // So far, so good!
        System.out.println("handshake(): successfuly completed.\n");
        return(true);
    }


    public boolean connect() {

        JSONArray       jsonRequest;
        JSONObject      jsonConnect;
        String          request;

        // Build JSON connect request
        System.out.print("connect(): building JSON connect request... ");
        try {


            jsonConnect = new JSONObject()
                .put("channel","/meta/connect")
                .put("clientId", this.clientId)
                .put("connectionType", "streaming");

            jsonRequest = new JSONArray()
                .put(jsonConnect);

            request = jsonRequest.toString();
            System.out.println("done.\n\t" + request);

        } catch (JSONException e) {

            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);
        }

        // Post JSON connect request
        System.out.print("connect(): posting request... ");
        try {

            this.httpPost.setEntity(new StringEntity(request,"UTF-8"));
            this.httpResponse = this.httpClient.execute(this.httpPost);
            System.out.println("done.");

        } catch(UnsupportedEncodingException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(ClientProtocolException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IOException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        }

        // Get Cometd server's response
        System.out.print("connect(): dispatching response... ");
        try {
            this.inputStreamReader  = new InputStreamReader(this.httpResponse.getEntity().getContent(),"UTF-8");
            this.inputStreamHandler = new InputStreamHandler(this);
            this.inputStreamHandler.start();
            System.out.println("done.");

        } catch( IllegalStateException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IOException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        }

        // So far, so good
        System.out.println("connect(): successfuly completed.\n");
        return true;
    }


    public boolean subscribe(JSONObject jsonDATA) {
        JSONArray       jsonRequest;
        JSONObject      jsonSubscribe;
        String          request;

        // Build JSON connect request
        System.out.print("subscribe(): building JSON subscribe request... ");
        try {

            jsonSubscribe = new JSONObject()
                .put("channel","/slim/subscribe")
                .put("data", jsonDATA)
                .put("id", this.requestId++);

            jsonRequest = new JSONArray()
                .put(jsonSubscribe);

            request = jsonRequest.toString();
            System.out.println("done.\n\t" + request);

        } catch (JSONException e) {

            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);
        }

        // Post JSON subscribe request
        System.out.print("subscribe(): posting request... ");
        try {

            this.httpPost.setEntity(new StringEntity(request,"UTF-8"));
            this.httpClient.execute(this.httpPost);
            this.inputStreamReader  = new InputStreamReader(this.httpResponse.getEntity().getContent(),"UTF-8");
            System.out.println("done.");

        } catch(UnsupportedEncodingException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(ClientProtocolException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IllegalStateException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        } catch(IOException e) {
            System.out.println("FAILED!");
            e.printStackTrace();
            return(false);

        }

        // So far, so good
        System.out.println();
        return true;
    }


    class InputStreamHandler extends Thread {

        private CometClient         parent;

        public InputStreamHandler(CometClient parent) {
            this.parent = parent;
        }

        public void run() {

            StringBuilder   result = new StringBuilder();
            int n = 0;

            try {
                    while( n >= 0 ) {

                        char[] buffer = new char[1024];

                        n = this.parent.getInputStreamReader().read(buffer, 0, buffer.length);
                        if( n > 0 ) {
                            result.append(buffer,0,n);

                            if( buffer[n] == '\0' ) {
                                System.out.println("InputStreamHandler thread: got response :\n\t" + result);

                                // Parse response
                                JSONArray   jsonResponse;
                                System.out.print("InputStreamHandler(): parsing response... ");
                                try {
                                    jsonResponse = new JSONArray(result.toString());
                                    System.out.println("done.");

                                } catch ( JSONException e ) {
                                    System.out.println("FAILED!");
                                    e.printStackTrace();
                                    return;
                                }

                                // Extract objects
                                System.out.println("InputStreamHandler(): extracting " + jsonResponse.length() + " objects:");
                                for( int i=0; i<jsonResponse.length(); i++) {
                                    JSONObject jsonObject;
                                    try {
                                        jsonObject = jsonResponse.getJSONObject(i);
                                        System.out.println("  jsonObject["+i+"]: " + jsonObject.toString());

                                        String channel = jsonObject.getString("channel");
                                        if( channel.equals("/meta/connect") ) {
                                            if( jsonObject.getBoolean("successful") == true ) {
                                                System.out.println("InputStreamHandler(): connection successfully completed.");
                                                result.delete(0, result.length());
                                                System.out.println("InputStreamHandler(): subscribing to 'serverstatus' now...");
                                                JSONObject jsonDATA = new JSONObject("{\"request\":[\"\",[\"serverstatus\",0,50,\"subscribe:30\"]],\"response\":\"/" + this.parent.getClientId() + "/slim/serverstatus\"}");
                                                this.parent.subscribe(jsonDATA);
                                                System.out.println("InputStreamHandler(): back in town.");
                                            } else {
                                                System.out.println("InputStreamHandler(): connection failed!");
                                            }
                                        } else if( channel.equals("/slim/subscribe") ) {
                                            System.out.println("InputStreamHandler(): subscription update.");
                                        }
                                    } catch (JSONException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                    }
                                }


                            }
                        } else {
                            System.out.println("InputStreamHandler thread: input stream is NOT ready.");
                        }
                        Thread.sleep(1000);
                    }
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }

    }

}