summaryrefslogtreecommitdiff
path: root/src/io/StreamImpl.java
blob: 84d9d837a418a0d46916f86ed727189d2b1a8131 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
package io;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.net.URLEncoder;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;
import mining.Stream;
import oauth.signpost.exception.OAuthException;
import org.apache.commons.io.Charsets;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.json.JSONException;
import org.json.JSONObject;
import org.json.JSONTokener;
import provider.ExceptionListener;
import provider.ResultListener;
import support.StreamingGZIPInputStream;

/**
 * Provides access to Twitter stream data. Data is processed in a separate
 * thread and consumers are notified via callbacks.
 *
 * @author Peter Wu
 */
public class StreamImpl implements Stream {

    private final static String STREAM_URL
            = "https://stream.twitter.com/1.1/statuses/filter.json";

    private final Set<String> keywords = new HashSet<>();
    /**
     * Used for thread-safe modifications.
     */
    private final Object listenerSync = new Object();
    private final Object resultListenerSync = new Object();
    /**
     * The target that is interested in received tweets.
     */
    private ResultListener resultListener;
    /**
     * The target that is interesting in receiving changes.
     */
    private ExceptionListener exceptionListener;
    /**
     * Used for signing messages.
     */
    private final OAuthRequester oauth;

    /**
     * Holds an instance that represents an active worker that contains a thread
     * to watch for new tweets.
     */
    private WorkerContainer workerContainer;

    public StreamImpl(OAuthRequester oauth) {
        this.oauth = oauth;
    }

    /**
     * Sets the listener for new tweets.
     *
     * @param resultListener
     */
    public void setResultListener(ResultListener resultListener) {
        // sync to avoid breakage when the poller thread sends a notification
        synchronized (resultListenerSync) {
            this.resultListener = resultListener;
        }
    }

    public ResultListener getResultListener() {
        return resultListener;
    }

    public void setExceptionListener(ExceptionListener exceptionListener) {
        synchronized (listenerSync) {
            this.exceptionListener = exceptionListener;
        }
    }

    private Set<String> splitKeywords(String rawKeywords) {
        Set<String> filteredKeywords = new HashSet<>();
        List<String> keywordsList = Arrays.asList(rawKeywords.split(","));
        for (String keyword : keywordsList) {
            if (!keyword.isEmpty()) {
                filteredKeywords.add(keyword);
            }
        }
        return filteredKeywords;
    }

    @Override
    public void watchKeyword(String rawKeywords) {
        for (String keyword : splitKeywords(rawKeywords)) {
            keywords.add(keyword);
        }
    }

    @Override
    public void unwatchKeyword(String rawKeywords) {
        for (String keyword : splitKeywords(rawKeywords)) {
            keywords.remove(keyword);
        }
    }

    @Override
    public Set<String> getKeywords(boolean active) {
        HashSet<String> retKeywords = new HashSet<>();
        if (active) {
            // return keywords from the active connection
            if (workerContainer != null) {
                String keywordsStr = workerContainer.getWorker().getKeywords();
                retKeywords.addAll(Arrays.asList(keywordsStr.split(",")));
            }
        } else {
            retKeywords.addAll(keywords);
        }
        return retKeywords;
    }

    @Override
    public void commit() throws IOException {
        String keywordsStr = StringUtils.join(keywords, ",");
        /* do not reconnect if a connection already exists for the keywords */
        if (workerContainer != null
                && workerContainer.getWorker().getKeywords().equals(keywordsStr)) {
            return;
        }
        // a query is required.
        if (keywords.isEmpty()) {
            return;
        }
        /* connect (or reconnect) after setting new keywords */
        disconnect();
        Worker worker = new Worker(keywordsStr);
        workerContainer = new WorkerContainer(worker);
        workerContainer.start();
    }

    @Override
    public boolean isValid() {
        return workerContainer != null && workerContainer.isValid();
    }

    @Override
    public void close() {
        disconnect();
    }

    /**
     * Stops the worker if any.
     */
    private void disconnect() {
        // wait for the worker thread to stop
        while (workerContainer != null) {
            workerContainer.finish();
            workerContainer = null;
        }
    }

    /**
     * Groups an activated stream connection.
     */
    private class WorkerContainer {

        private final Worker worker;
        private final Poller poller;
        private final Thread ioThread;
        private final Thread pollerThread;

        /**
         * Holds a thread for the worker.
         *
         * @param worker Worker that should be processed in a separate thread.
         */
        public WorkerContainer(Worker worker) {
            this.worker = worker;
            this.poller = new Poller(worker);
            ioThread = new Thread(worker);
            // the poller is closely coupled with the I/O thread. If the I/O
            // thread dies, no new messages will be added to the queue.
            pollerThread = new Thread(poller);
        }

        public void start() {
            ioThread.start();
            pollerThread.start();
        }

        public void finish() {
            worker.stopWorker();
            while (ioThread.isAlive()) {
                try {
                    ioThread.join();
                    // maybe the poller is stuck waiting for a new object, wake
                    // it up.
                    pollerThread.interrupt();
                    pollerThread.join();
                } catch (InterruptedException ex) {
                    Logger.getLogger(getClass().getName())
                            .warning("Interrupted while waiting for stream finish");
                }
            }
        }

        public Worker getWorker() {
            return worker;
        }

        public boolean isValid() {
            return ioThread.isAlive();
        }
    }

    private class Poller implements Runnable {

        private final Worker worker;

        private Poller(Worker worker) {
            this.worker = worker;
        }

        @Override
        public void run() {
            // keep waiting for objects if the worker is alive, or fetch objects
            // if the worker has any old ones left.
            while (worker.isRunning() || worker.hasObjects()) {
                try {
                    JSONObject obj = worker.getObject();
                    processObject(obj);
                } catch (InterruptedException ex) {
                    // interrupted, probably signalled to stop?
                }
            }
        }

        private void processObject(JSONObject obj) {
            try {
                JSONObject user = obj.getJSONObject("user");
                resultListener.profileGenerated(user);
            } catch (JSONException ex) {
                // should not happen because the worker inserts tweets (which
                // assumes that a tweet has a user member).
                Logger.getLogger(getClass().getName())
                        .severe("Expected a user in a tweet!");
            }
            synchronized (resultListenerSync) {
                if (resultListener != null) {
                    resultListener.tweetGenerated(obj);
                }
            }
        }
    }

    private class Worker implements Runnable {

        private final String keywords;
        private final HttpURLConnection connection;
        private volatile boolean running = true;
        private final BlockingQueue<JSONObject> receivedObjects;
        private InputStream inputStream;

        Worker(String keywords) throws IOException {
            this.keywords = keywords;
            try {
                this.connection = connect(keywords);
                this.inputStream = this.connection.getInputStream();
            } catch (IOException ex) {
                IOUtils.closeQuietly(this.inputStream);
                if (this.connection != null) {
                    this.connection.disconnect();
                }
                throw ex;
            }
            this.receivedObjects = new LinkedBlockingQueue<>();
        }

        /**
         * @return The search keywords associated with this worker.
         */
        public String getKeywords() {
            return keywords;
        }

        private HttpURLConnection connect(String keywords) throws IOException {
            String postData = "track=" + URLEncoder.encode(keywords, "UTF-8");
            postData += "&language=en";
            HttpURLConnection conn;
            conn = (HttpURLConnection) new URL(STREAM_URL).openConnection();
            conn.setRequestMethod("POST");
            // set request headers
            conn.addRequestProperty("Content-Type",
                    "application/x-www-form-urlencoded; charset=UTF-8");
            conn.addRequestProperty("Accept-Encoding", "gzip");
            // connect and send request
            conn.setDoOutput(true);
            try {
                oauth.sign(conn, postData);
            } catch (OAuthException ex) {
                throw new IOException("Unable to sign request", ex);
            }
            conn.getOutputStream().write(postData.getBytes(Charsets.UTF_8));
            int respCode = conn.getResponseCode();
            if (respCode != 200) {
                getLogger().severe("Response code " + respCode);
                conn.disconnect();
                throw new IOException("Unexpected stream response " + respCode);
            }
            return conn;
        }

        /**
         * Wraps an inputstream as gzip if possible.
         *
         * @param is The raw input stream.
         * @return An inputstream that outputs decoded data.
         * @throws IOException
         */
        private InputStream wrapGzip(InputStream is) throws IOException {
            if ("gzip".equals(connection.getContentEncoding())) {
                return new StreamingGZIPInputStream(is);
            }
            return is;
        }

        @Override
        public void run() {
            InputStream is = inputStream;
            IOException run_error = null;
            try {
                parseMainLoop(wrapGzip(is));
            } catch (IOException ex) {
                run_error = ex;
            } finally {
                IOUtils.closeQuietly(is);
                connection.disconnect();
                if (run_error != null) {
                    // synchronize just in case the exception listener gets
                    // modified while hell breaks lose.
                    synchronized (listenerSync) {
                        if (exceptionListener != null) {
                            exceptionListener.exceptionGenerated(run_error);
                        }
                    }
                }
            }
        }

        public void stopWorker() {
            /* inform the worker to stop as soon as possible */
            running = false;
            try {
                inputStream.close();
            } catch (IOException ex) {
                getLogger().warning("Error while closing stream: "
                        + ex.getMessage());
            }
            connection.disconnect();
        }

        public boolean isRunning() {
            return running;
        }

        private void parseMainLoop(InputStream is) throws IOException {
            // Note: one message per CRLF-terminated line
            // See https://dev.twitter.com/docs/streaming-apis/messages
            InputStreamReader isr = new InputStreamReader(is, Charsets.UTF_8);
            BufferedReader reader = new BufferedReader(isr);
            JSONTokener jsonTokener = new JSONTokener(reader);
            while (running) {
                try {
                    Object obj = jsonTokener.nextValue();
                    if (obj instanceof JSONObject) {
                        processReceivedObject((JSONObject) obj);
                    } else {
                        getLogger().severe("Got unexpected object: " + obj);
                        throw new IOException("Got unexpected type from stream");
                    }
                } catch (JSONException ex) {
                    // ignore IO errors for a stop request ("Socket closed")
                    if (running) {
                        throw new IOException(ex);
                    }
                }
            }
        }

        /**
         * Detect tweets and queue them for processing.
         *
         * @param obj an object received at the stream.
         */
        private void processReceivedObject(JSONObject obj) {
            // assume that tweets always have a user field
            if (obj.has("user")) {
                receivedObjects.offer(obj);
            } else {
                getLogger().warning("Received unknown object: " + obj);
            }
        }

        public boolean hasObjects() {
            return !receivedObjects.isEmpty();
        }

        public JSONObject getObject() throws InterruptedException {
            return receivedObjects.take();
        }

        private Logger getLogger() {
            return Logger.getLogger(getClass().getName());
        }
    }
}