Step 4: Add the Source Code

1swfobject.registerObject("clippy.codeblock-0", "9");
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17package demo;
18
19import org.cometd.bayeux.Channel;
20import org.cometd.bayeux.Message;
21import org.cometd.bayeux.client.ClientSessionChannel;
22import org.cometd.bayeux.client.ClientSessionChannel.MessageListener;
23import org.cometd.client.BayeuxClient;
24import org.cometd.client.transport.ClientTransport;
25import org.cometd.client.transport.LongPollingTransport;
26
27import org.eclipse.jetty.client.ContentExchange;
28import org.eclipse.jetty.client.HttpClient;
29
30import java.net.MalformedURLException;
31import java.net.URL;
32import java.util.HashMap;
33import java.util.Map;
34
35/**
36 * This example demonstrates how a streaming client works 
37 * against the Salesforce Streaming API.
38 **/
39
40public class StreamingClientExample {
41
42    // This URL is used only for logging in. The LoginResult 
43    // returns a serverUrl which is then used for constructing
44    // the streaming URL. The serverUrl points to the endpoint 
45    // where your organization is hosted.
46
47    static final String LOGIN_ENDPOINT = "https://login.salesforce.com";
48    private static final String USER_NAME = "change_this_to_your_testuser@yourcompany.com";
49    private static final String PASSWORD = "change_this_to_your_testpassword"; 
50    // NOTE: Putting passwords in code is not a good practice and not recommended.
51
52    // Set this to true only when using this client 
53    // against the Summer'11 release (API version=22.0).
54    private static final boolean VERSION_22 = false;  
55    private static final boolean USE_COOKIES = VERSION_22;
56
57    // The channel to subscribe to. Same as the name of the PushTopic. 
58    // Be sure to create this topic before running this sample.
59    private static final String CHANNEL = VERSION_22 ? "/InvoiceStatementUpdates" : "/topic/InvoiceStatementUpdates";
60    private static final String STREAMING_ENDPOINT_URI = VERSION_22 ? 
61        "/cometd" : "/cometd/34.0";
62
63    // The long poll duration.
64    private static final int CONNECTION_TIMEOUT = 20 * 1000;  // milliseconds
65    private static final int READ_TIMEOUT = 120 * 1000; // milliseconds
66
67    public static void main(String[] args) throws Exception {
68
69    	System.out.println("Running streaming client example....");
70
71        final BayeuxClient client = makeClient();
72        client.getChannel(Channel.META_HANDSHAKE).addListener
73            (new ClientSessionChannel.MessageListener() {
74
75            public void onMessage(ClientSessionChannel channel, Message message) {
76
77            	System.out.println("[CHANNEL:META_HANDSHAKE]: " + message);
78
79                boolean success = message.isSuccessful();
80                if (!success) {
81                    String error = (String) message.get("error");
82                    if (error != null) {
83                        System.out.println("Error during HANDSHAKE: " + error);
84                        System.out.println("Exiting...");
85                        System.exit(1);
86                    }
87
88                    Exception exception = (Exception) message.get("exception");
89                    if (exception != null) {
90                        System.out.println("Exception during HANDSHAKE: ");
91                        exception.printStackTrace();
92                        System.out.println("Exiting...");
93                        System.exit(1);
94
95                    }
96                }
97            }
98
99        });
100
101        client.getChannel(Channel.META_CONNECT).addListener(
102            new ClientSessionChannel.MessageListener() {
103            public void onMessage(ClientSessionChannel channel, Message message) {
104
105                System.out.println("[CHANNEL:META_CONNECT]: " + message);
106
107                boolean success = message.isSuccessful();
108                if (!success) {
109                    String error = (String) message.get("error");
110                    if (error != null) {
111                        System.out.println("Error during CONNECT: " + error);
112                        System.out.println("Exiting...");
113                        System.exit(1);
114                    }
115                }
116            }
117
118        });
119
120        client.getChannel(Channel.META_SUBSCRIBE).addListener(
121            new ClientSessionChannel.MessageListener() {
122
123            public void onMessage(ClientSessionChannel channel, Message message) {
124
125            	System.out.println("[CHANNEL:META_SUBSCRIBE]: " + message);
126                boolean success = message.isSuccessful();
127                if (!success) {
128                    String error = (String) message.get("error");
129                    if (error != null) {
130                        System.out.println("Error during SUBSCRIBE: " + error);
131                        System.out.println("Exiting...");
132                        System.exit(1);
133                    }
134                }
135            }
136        });
137
138
139
140        client.handshake();
141        System.out.println("Waiting for handshake");
142        
143        boolean handshaken = client.waitFor(10 * 1000, BayeuxClient.State.CONNECTED);
144        if (!handshaken) {
145            System.out.println("Failed to handshake: " + client);
146            System.exit(1);
147        }
148
149
150
151        System.out.println("Subscribing for channel: " + CHANNEL);
152
153        client.getChannel(CHANNEL).subscribe(new MessageListener() {
154            @Override
155            public void onMessage(ClientSessionChannel channel, Message message) {
156                System.out.println("Received Message: " + message);
157            }
158        });
159
160        System.out.println("Waiting for streamed data from your organization ...");
161        while (true) {
162            // This infinite loop is for demo only,
163            // to receive streamed events on the 
164            // specified topic from your organization.
165        }
166    }
167
168
169
170    private static BayeuxClient makeClient() throws Exception {
171        HttpClient httpClient = new HttpClient();
172        httpClient.setConnectTimeout(CONNECTION_TIMEOUT);
173        httpClient.setTimeout(READ_TIMEOUT);
174        httpClient.start();
175
176        String[] pair = SoapLoginUtil.login(httpClient, USER_NAME, PASSWORD);
177        
178        if (pair == null) {
179            System.exit(1);
180        }
181
182        assert pair.length == 2;
183        final String sessionid = pair[0];
184        String endpoint = pair[1];
185        System.out.println("Login successful!\nEndpoint: " + endpoint 
186            + "\nSessionid=" + sessionid);
187
188        Map<String, Object> options = new HashMap<String, Object>();
189        options.put(ClientTransport.TIMEOUT_OPTION, READ_TIMEOUT);
190        LongPollingTransport transport = new LongPollingTransport(
191          options, httpClient) {
192
193        	@Override
194            protected void customize(ContentExchange exchange) {
195                super.customize(exchange);
196                exchange.addRequestHeader("Authorization", "OAuth " + sessionid);
197            }
198        };
199
200        BayeuxClient client = new BayeuxClient(salesforceStreamingEndpoint(
201            endpoint), transport);
202        if (USE_COOKIES) establishCookies(client, USER_NAME, sessionid);
203        return client;
204    }
205
206    private static String salesforceStreamingEndpoint(String endpoint)
207        throws MalformedURLException {
208        return new URL(endpoint + STREAMING_ENDPOINT_URI).toExternalForm();
209    }
210
211    private static void establishCookies(BayeuxClient client, String user, 
212        String sid) {
213        client.setCookie("com.salesforce.LocaleInfo", "us", 24 * 60 * 60 * 1000);
214        client.setCookie("login", user, 24 * 60 * 60 * 1000);
215        client.setCookie("sid", sid, 24 * 60 * 60 * 1000);
216        client.setCookie("language", "en_US", 24 * 60 * 60 * 1000);
217    }
218}
219
220

1swfobject.registerObject("clippy.codeblock-1", "9");
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17package demo;
18
19import java.io.ByteArrayInputStream;
20import java.io.IOException;
21import java.io.UnsupportedEncodingException;
22import java.net.MalformedURLException;
23import java.net.URL;
24
25import org.eclipse.jetty.client.ContentExchange;
26import org.eclipse.jetty.client.HttpClient;
27import org.xml.sax.Attributes;
28import org.xml.sax.SAXException;
29import org.xml.sax.helpers.DefaultHandler;
30
31import javax.xml.parsers.ParserConfigurationException;
32import javax.xml.parsers.SAXParser;
33import javax.xml.parsers.SAXParserFactory;
34
35public final class SoapLoginUtil {
36
37    // The enterprise SOAP API endpoint used for the login call in this example.
38    private static final String SERVICES_SOAP_PARTNER_ENDPOINT = "/services/Soap/u/22.0/";
39
40    private static final String ENV_START =
41            "<soapenv:Envelope xmlns:soapenv='http://schemas.xmlsoap.org/soap/envelope/' "
42                    + "xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' " +
43                    "xmlns:urn='urn:partner.soap.sforce.com'><soapenv:Body>";
44
45    private static final String ENV_END = "</soapenv:Body></soapenv:Envelope>";
46
47    private static byte[] soapXmlForLogin(String username, String password) 
48        throws UnsupportedEncodingException {
49        return (ENV_START +
50                "  <urn:login>" +
51                "    <urn:username>" + username + "</urn:username>" +
52                "    <urn:password>" + password + "</urn:password>" +
53                "  </urn:login>" +
54                ENV_END).getBytes("UTF-8");
55    }
56
57    public static String[] login(HttpClient client, String username, String password)
58            throws IOException, InterruptedException, SAXException, 
59                ParserConfigurationException {
60
61        ContentExchange exchange = new ContentExchange();
62        exchange.setMethod("POST");
63        exchange.setURL(getSoapURL());
64        exchange.setRequestContentSource(new ByteArrayInputStream(soapXmlForLogin(
65            username, password)));
66        exchange.setRequestHeader("Content-Type", "text/xml");
67        exchange.setRequestHeader("SOAPAction", "''");
68        exchange.setRequestHeader("PrettyPrint", "Yes");
69
70        client.send(exchange);
71        exchange.waitForDone();
72        String response = exchange.getResponseContent();
73
74        SAXParserFactory spf = SAXParserFactory.newInstance();
75        spf.setNamespaceAware(true);
76        SAXParser saxParser = spf.newSAXParser();
77
78        LoginResponseParser parser = new LoginResponseParser();
79        saxParser.parse(new ByteArrayInputStream(
80            response.getBytes("UTF-8")), parser);
81
82        if (parser.sessionId == null || parser.serverUrl == null) {
83            System.out.println("Login Failed!\n" + response);
84            return null;
85        }
86
87        URL soapEndpoint = new URL(parser.serverUrl);
88        StringBuilder endpoint = new StringBuilder()
89            .append(soapEndpoint.getProtocol())
90            .append("://")
91            .append(soapEndpoint.getHost());
92
93        if (soapEndpoint.getPort() > 0) endpoint.append(":")
94            .append(soapEndpoint.getPort());
95        return new String[] {parser.sessionId, endpoint.toString()};
96    }
97
98    private static String getSoapURL() throws MalformedURLException {
99        return new URL(StreamingClientExample.LOGIN_ENDPOINT + 
100            getSoapUri()).toExternalForm();
101    }
102
103    private static String getSoapUri() {
104        return SERVICES_SOAP_PARTNER_ENDPOINT;
105    }
106
107    private static class LoginResponseParser extends DefaultHandler {
108
109        private boolean inSessionId;
110        private String sessionId;
111
112        private boolean inServerUrl;
113        private String serverUrl;
114
115        @Override
116        public void characters(char[] ch, int start, int length) {
117            if (inSessionId) sessionId = new String(ch, start, length);
118            if (inServerUrl) serverUrl = new String(ch, start, length);
119        }
120
121        @Override
122        public void endElement(String uri, String localName, String qName) {
123            if (localName != null) {
124                if (localName.equals("sessionId")) {
125                    inSessionId = false;
126                }
127                
128                if (localName.equals("serverUrl")) {
129                    inServerUrl = false;
130                }
131            }
132        }
133
134        @Override
135        public void startElement(String uri, String localName, 
136            String qName, Attributes attributes) {
137            if (localName != null) {
138                if (localName.equals("sessionId")) {
139                    inSessionId = true;
140                }
141
142                if (localName.equals("serverUrl")) {
143                    inServerUrl = true;
144                }
145            }
146        }
147    }
148}