Implemented initial notification API support.

Fixes issue hub4j#119

src/main/java/org/kohsuke/github/GHNotificationStream.java

@@ -0,0 +1,193 @@
+package org.kohsuke.github;
+
+import java.io.IOException;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * Listens to GitHub notification stream.
+ *
+ * <p>
+ * This class supports two modes of retrieving notifications that can
+ * be controlled via {@link #nonBlocking(boolean)}.
+ *
+ * <p>
+ * In the blocking mode, which is the default, iterator will be infinite.
+ * The call to {@link Iterator#next()} will block until a new notification
+ * arrives. This is useful for application that runs perpetually and reacts
+ * to notifications.
+ *
+ * <p>
+ * In the non-blocking mode, the iterator will only report the set of
+ * notifications initially retrieved from GitHub, then quit. This is useful
+ * for a batch application to process the current set of notifications.
+ *
+ * @author Kohsuke Kawaguchi
+ * @see GitHub#listNotifications()
+ * @see GHRepository#listNotifications()
+ */
+public class GHNotificationStream implements Iterable<GHThread> {
+    private final GitHub root;
+
+    private Boolean all, participating;
+    private String since;
+    private String apiUrl;
+    private boolean nonBlocking = false;
+
+    /*package*/ GHNotificationStream(GitHub root, String apiUrl) {
+        this.root = root;
+        this.apiUrl = apiUrl;
+    }
+
+    /**
+     * Should the stream include notifications that are already read?
+     */
+    public GHNotificationStream read(boolean v) {
+        all = v;
+        return this;
+    }
+
+    /**
+     * Should the stream be restricted to notifications in which the user
+     * is directly participating or mentioned?
+     */
+    public GHNotificationStream participating(boolean v) {
+        participating = v;
+        return this;
+    }
+
+    public GHNotificationStream since(long timestamp) {
+        return since(new Date(timestamp));
+    }
+
+    public GHNotificationStream since(Date dt) {
+        since = GitHub.printDate(dt);
+        return this;
+    }
+
+    /**
+     * If set to true, {@link #iterator()} will stop iterating instead of blocking and
+     * waiting for the updates to arrive.
+     */
+    public GHNotificationStream nonBlocking(boolean v) {
+        this.nonBlocking = v;
+        return this;
+    }
+
+    /**
+     * Returns an infinite blocking {@link Iterator} that returns
+     * {@link GHThread} as notifications arrive.
+     */
+    public Iterator<GHThread> iterator() {
+        // capture the configuration setting here
+        final Requester req = new Requester(root).method("GET")
+                .with("all", all).with("participating", participating).with("since", since);
+
+        return new Iterator<GHThread>() {
+            /**
+             * Stuff we've fetched but haven't returned to the caller.
+             * Newer ones first.
+             */
+            private GHThread[] threads = EMPTY_ARRAY;
+
+            /**
+             * Next element in {@link #threads} to return. This counts down.
+             */
+            private int idx=-1;
+
+            /**
+             * threads whose updated_at is older than this should be ignored.
+             */
+            private long lastUpdated = -1;
+
+            /**
+             * Next request should have "If-Modified-Since" header with this value.
+             */
+            private String lastModified;
+
+            /**
+             * When is the next polling allowed?
+             */
+            private long nextCheckTime = -1;
+
+            private GHThread next;
+
+            public GHThread next() {
+                if (next==null) {
+                    next = fetch();
+                    if (next==null)
+                        throw new NoSuchElementException();
+                }
+
+                GHThread r = next;
+                next = null;
+                return r;
+            }
+
+            public boolean hasNext() {
+                if (next==null)
+                    next = fetch();
+                return next!=null;
+            }
+
+            GHThread fetch() {
+                try {
+                    while (true) {// loop until we get new threads to return
+
+                        // if we have fetched un-returned threads, use them first
+                        while (idx>=0) {
+                            GHThread n = threads[idx--];
+                            long nt = n.getUpdatedAt().getTime();
+                            if (nt >= lastUpdated) {
+                                lastUpdated = nt;
+                                return n;
+                            }
+                        }
+
+                        if (nonBlocking && nextCheckTime>=0)
+                            return null;    // nothing more to report, and we aren't blocking
+
+                        // observe the polling interval before making the call
+                        while (true) {
+                            long now = System.currentTimeMillis();
+                            if (nextCheckTime < now) break;
+                            long waitTime = Math.max(Math.min(nextCheckTime - now, 1000), 60 * 1000);
+                            Thread.sleep(waitTime);
+                        }
+
+                        req.setHeader("If-Modified-Since", lastModified);
+
+                        threads = req.to(apiUrl, GHThread[].class);
+                        if (threads==null) {
+                            threads = EMPTY_ARRAY;  // if unmodified, we get empty array
+                        } else {
+                            // we get a new batch, but we want to ignore the ones that we've seen
+                            lastUpdated++;
+                        }
+                        idx = threads.length-1;
+
+                        nextCheckTime = calcNextCheckTime();
+                        lastModified = req.getResponseHeader("Last-Modified");
+                    }
+                } catch (IOException e) {
+                    throw new RuntimeException(e);
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+            }
+
+            private long calcNextCheckTime() {
+                String v = req.getResponseHeader("X-Poll-Interval");
+                if (v==null)    v="60";
+                return System.currentTimeMillis()+Integer.parseInt(v)*1000;
+            }
+
+            public void remove() {
+                throw new UnsupportedOperationException();
+            }
+        };
+    }
+
+    private static final GHThread[] EMPTY_ARRAY = new GHThread[0];
+}

src/main/java/org/kohsuke/github/GHRepository.java

@@ -1170,7 +1170,13 @@ public Reader renderMarkdown(String text, MarkdownMode mode) throws IOException
             "UTF-8");
     }
 
-
+    /**
+     * List all the notifications in a repository for the current user.
+     */
+    public GHNotificationStream listNotifications() {
+        return new GHNotificationStream(root,getApiTailUrl("/notifications"));
+    }
+
 
     @Override
     public String toString() {

src/main/java/org/kohsuke/github/GHThread.java

@@ -0,0 +1,57 @@
+package org.kohsuke.github;
+
+import java.util.Date;
+
+/**
+ * A conversation in the notification API.
+ *
+ *
+ * @see <a href="https://developer.github.com/v3/activity/notifications/">documentation</a>
+ * @author Kohsuke Kawaguchi
+ */
+public class GHThread extends GHObject {
+    private GHRepository repository;
+    private Subject subject;
+    private String reason;
+    private boolean unread;
+    private String last_read_at;
+
+    static class Subject {
+        String title;
+        String url;
+        String latest_comment_url;
+        String type;
+    }
+
+    private GHThread() {// no external construction allowed
+    }
+
+    /**
+     * Returns null if the entire thread has never been read.
+     */
+    public Date getLastReadAt() {
+        return GitHub.parseDate(last_read_at);
+    }
+
+    public String getReason() {
+        return reason;
+    }
+
+    public GHRepository getRepository() {
+        return repository;
+    }
+
+    // TODO: how to expose the subject?
+
+    public boolean isRead() {
+        return !unread;
+    }
+
+    public String getTitle() {
+        return subject.title;
+    }
+
+    public String getType() {
+        return subject.type;
+    }
+}

src/main/java/org/kohsuke/github/GitHub.java

@@ -464,6 +464,13 @@ public GHContentSearchBuilder searchContent() {
         return new GHContentSearchBuilder(this);
     }
 
+    /**
+     * List all the notifications.
+     */
+    public GHNotificationStream listNotifications() {
+        return new GHNotificationStream(this,"/notifications");
+    }
+
     /**
      * This provides a dump of every public repository, in the order that they were created.
      * @see <a href="https://developer.github.com/v3/repos/#list-all-public-repositories">documentation</a>

src/main/java/org/kohsuke/github/Requester.java

@@ -43,8 +43,10 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.NoSuchElementException;
 import java.util.Set;
 import java.util.regex.Matcher;
@@ -62,6 +64,7 @@
 class Requester {
     private final GitHub root;
     private final List<Entry> args = new ArrayList<Entry>();
+    private final Map<String,String> headers = new LinkedHashMap<String, String>();
 
     /**
      * Request method.
@@ -89,6 +92,15 @@ private Entry(String key, Object value) {
         this.root = root;
     }
 
+    /**
+     * Sets the request HTTP header.
+     *
+     * If a header of the same name is already set, this method overrides it.
+     */
+    public void setHeader(String name, String value) {
+        headers.put(name,value);
+    }
+
     /**
      * Makes a request with authentication credential.
      */
@@ -267,6 +279,11 @@ public InputStream read(String tailApiUrl) throws IOException {
         }
     }
 
+    public String getResponseHeader(String header) {
+        return uc.getHeaderField(header);
+    }
+
+
     /**
      * Set up the request parameters or POST payload.
      */
@@ -406,6 +423,12 @@ private void setupConnection(URL url) throws IOException {
         if (root.encodedAuthorization!=null)
             uc.setRequestProperty("Authorization", root.encodedAuthorization);
 
+        for (Map.Entry<String, String> e : headers.entrySet()) {
+            String v = e.getValue();
+            if (v!=null)
+                uc.setRequestProperty(e.getKey(), v);
+        }
+
         try {
             uc.setRequestMethod(method);
         } catch (ProtocolException e) {
@@ -422,6 +445,8 @@ private void setupConnection(URL url) throws IOException {
     }
 
     private <T> T parse(Class<T> type, T instance) throws IOException {
+        if (uc.getResponseCode()==304)
+            return null;    // special case handling for 304 unmodified, as the content will be ""
         InputStreamReader r = null;
         try {
             r = new InputStreamReader(wrapStream(uc.getInputStream()), "UTF-8");

src/test/java/org/kohsuke/github/AppTest.java

@@ -807,6 +807,22 @@ public void searchContent() throws Exception {
         assertTrue(r.getTotalCount() > 0);
     }
 
+    @Test
+    public void notifications() throws Exception {
+        boolean found=false;
+        for (GHThread t : gitHub.listNotifications().nonBlocking(true)) {
+            found = true;
+            assertNotNull(t.getTitle());
+            assertNotNull(t.getReason());
+
+            System.out.println(t.getTitle());
+            System.out.println(t.getLastReadAt());
+            System.out.println(t.getType());
+            System.out.println();
+        }
+        assertTrue(found);
+    }
+
     private void kohsuke() {
         String login = getUser().getLogin();
         Assume.assumeTrue(login.equals("kohsuke") || login.equals("kohsuke2"));