Merge pull request #16 from pimanac/streams

add ListingStream

Author: CrustyJew

README.md

@@ -48,6 +48,34 @@ foreach (var post in all.Take(25))
 }
 ```
 
+
+**Using ListingStreams**
+
+Use ListingStreams to infinitely yeild new Things posted to reddit
+
+Example:
+
+```csharp
+// get all new comments as they are posted.
+foreach (var comment in subreddit.CommentStream)
+{
+    Console.WriteLine(DateTime.Now + "   New Comment posted to /r/example: " + comment.ShortLink);
+}
+```
+
+you can call .GetListingStream() on any Listing<Thing>
+
+```csharp
+// get new modmail
+var newModmail = user.ModMail.GetListingStream();
+foreach (var message in newModmail)
+{
+    if (message.FirstMessageName == "")
+        message.Reply("Thanks for the message - we will get back to you soon.");
+}
+
+```
+
 ## Development
 
 RedditSharp is developed with the following workflow:

RedditSharp/Listing.cs

@@ -55,9 +55,9 @@ internal Listing(Reddit reddit, string url, IWebAgent webAgent)
         /// <param name="limitPerRequest">The number of listings to be returned per request</param>
         /// <param name="maximumLimit">The maximum number of listings to return</param>
         /// <returns></returns>
-        public IEnumerator<T> GetEnumerator(int limitPerRequest, int maximumLimit = -1)
+        public IEnumerator<T> GetEnumerator(int limitPerRequest, int maximumLimit = -1, bool stream = false)
         {
-            return new ListingEnumerator<T>(this, limitPerRequest, maximumLimit);
+            return new ListingEnumerator<T>(this, limitPerRequest, maximumLimit, stream);
         }
 
         /// <summary>
@@ -69,6 +69,7 @@ public IEnumerator<T> GetEnumerator()
             return GetEnumerator(DefaultListingPerRequest);
         }
 
+
         /// <summary>
         /// Returns an enumerator that iterates through a collection
         /// </summary>
@@ -103,6 +104,24 @@ public IEnumerable<T> GetListing(int maximumLimit, int limitPerRequest)
             return GetEnumerator(enumerator);
         }
 
+        /// <summary>
+        /// Returns an IEnumerable instance which will infinitely yield new <see cref="Thing"/> 
+        /// </summary>
+        /// <param name="limitPerRequest">
+        ///   Number of records to return in each request to the reddit api.  Defaults to using the reddit
+        ///   standard of 25 records of requests.
+        ///   Adjusting this up or down based on the size of your subreddit and the rate at which new content
+        ///   is created.
+        /// </param>
+        /// <param name="maximumLimit">maximum number of records to return</param>
+        /// <returns></returns>
+        public IEnumerable<T> GetListingStream(int limitPerRequest = -1, int maximumLimit = -1)
+        {
+            // Get the enumerator with the specified maximum and per request limits
+            var enumerator = GetEnumerator(limitPerRequest, maximumLimit, true);
+            return GetEnumerator(enumerator);
+        }
+
         /// <summary>
         /// Converts an IEnumerator instance to an IEnumerable
         /// </summary>
@@ -119,6 +138,7 @@ private static IEnumerable<T> GetEnumerator(IEnumerator<T> enumerator)
 #pragma warning disable 0693
         private class ListingEnumerator<T> : IEnumerator<T> where T : Thing
         {
+            private bool stream = false;
             private Listing<T> Listing { get; set; }
             private int CurrentPageIndex { get; set; }
             private string After { get; set; }
@@ -128,17 +148,22 @@ private class ListingEnumerator<T> : IEnumerator<T> where T : Thing
             private int LimitPerRequest { get; set; }
             private int MaximumLimit { get; set; }
 
+            private List<string> done;
+
             /// <summary>
             /// Creates a new ListingEnumerator instance
             /// </summary>
             /// <param name="listing"></param>
             /// <param name="limitPerRequest">The number of listings to be returned per request. -1 will exclude this parameter and use the Reddit default (25)</param>
             /// <param name="maximumLimit">The maximum number of listings to return, -1 will not add a limit</param>
-            public ListingEnumerator(Listing<T> listing, int limitPerRequest, int maximumLimit)
+            /// <param name="stream">yield new <see cref="Thing"/> as they are created</param>
+            public ListingEnumerator(Listing<T> listing, int limitPerRequest, int maximumLimit, bool stream = false)
             {
                 Listing = listing;
                 CurrentPageIndex = -1;
                 CurrentPage = new Thing[0];
+                done = new List<string>();
+                this.stream = stream;
 
                 // Set the listings per page (if not specified, use the Reddit default of 25) and the maximum listings
                 LimitPerRequest = (limitPerRequest <= 0 ? DefaultListingPerRequest : limitPerRequest); 
@@ -147,13 +172,24 @@ public ListingEnumerator(Listing<T> listing, int limitPerRequest, int maximumLim
 
             public T Current
             {
-                get 
+                get
                 {
                     return (T)CurrentPage[CurrentPageIndex];
                 }
             }
 
             private void FetchNextPage()
+            {
+                if (stream)
+                    PageForward();
+                else
+                    PageBack();
+            }
+
+            /// <summary>
+            /// Standard behavior.  Page from newest to oldest - "backward" in time.
+            /// </summary>
+            private void PageBack()
             {
                 var url = Listing.Url;
 
@@ -200,14 +236,83 @@ private void FetchNextPage()
                 Parse(json);
             }
 
+
+            /// <summary>
+            /// Page from oldest to newest - "forward" in time.
+            /// </summary>
+            private void PageForward()
+            {
+                var url = Listing.Url;
+
+                if (Before != null)
+                {
+                    url += (url.Contains("?") ? "&" : "?") + "before=" + Before;
+                }
+
+                if (LimitPerRequest != -1)
+                {
+                    int limit = LimitPerRequest;
+
+                    if (limit > MaximumLimit)
+                    {
+                        // If the limit is more than the maximum number of listings, adjust
+                        limit = MaximumLimit;
+                    }
+                    else if (Count + limit > MaximumLimit)
+                    {
+                        // If a smaller subset of listings are needed, adjust the limit
+                        limit = MaximumLimit - Count;
+                    }
+
+                    if (limit > 0)
+                    {
+                        // Add the limit, the maximum number of items to be returned per page
+                        url += (url.Contains("?") ? "&" : "?") + "limit=" + limit;
+                    }
+                }
+
+                if (Count > 0)
+                {
+                    // Add the count, the number of items already seen in this listingStream
+                    // The Reddit API uses this to determine when to give values for before and after fields                
+                    url += (url.Contains("?") ? "&" : "?") + "count=" + Count;
+                }
+
+                var request = Listing.WebAgent.CreateGet(url);
+                var response = request.GetResponse();
+                var data = Listing.WebAgent.GetResponseString(response.GetResponseStream());
+                var json = JToken.Parse(data);
+                if (json["kind"].ValueOrDefault<string>() != "Listing")
+                    throw new FormatException("Reddit responded with an object that is not a listingStream.");
+                Parse(json);
+            }
+
+
             private void Parse(JToken json)
             {
                 var children = json["data"]["children"] as JArray;
-                CurrentPage = new Thing[children.Count];
-                
-                for (int i = 0; i < CurrentPage.Length; i++)
-                    CurrentPage[i] = Thing.Parse<T>(Listing.Reddit, children[i], Listing.WebAgent);
+                var things = new List<Thing>();
 
+                for (int i = 0; i < children.Count; i++)
+                {
+                    if (!stream)
+                        things.Add(Thing.Parse<T>(Listing.Reddit, children[i], Listing.WebAgent));
+                    else
+                    {
+                        // we only want to see new items.
+                        var id = children[i]["data"]["id"].ValueOrDefault<string>();
+                        if (String.IsNullOrEmpty(id) || done.Contains(id))
+                            continue;
+
+                        things.Add(Thing.Parse<T>(Listing.Reddit, children[i], Listing.WebAgent));
+                        done.Add(id);
+                    }
+                }
+
+                if (stream)
+                    things.Reverse();
+
+                CurrentPage = things.ToArray();
                 // Increase the total count of items returned
                 Count += CurrentPage.Length;
 
@@ -226,6 +331,14 @@ object System.Collections.IEnumerator.Current
             }
 
             public bool MoveNext()
+            {
+                if (stream)
+                    return MoveNextForward();
+                else
+                    return MoveNextBack();
+            }
+
+            private bool MoveNextBack()
             {
                 CurrentPageIndex++;
                 if (CurrentPageIndex == CurrentPage.Length)
@@ -255,6 +368,63 @@ public bool MoveNext()
                 return true;
             }
 
+            private bool MoveNextForward()
+            {
+                CurrentPageIndex++;
+                if (CurrentPageIndex == CurrentPage.Length)
+                {
+                    int tries = 0;
+                    while (true)
+                    {
+                        if (MaximumLimit != -1 && Count >= MaximumLimit)
+                            return false;
+
+                        tries++;
+                        // Get the next page
+                        try
+                        {
+                            FetchNextPage();
+                        }
+                        catch (Exception ex)
+                        {
+                            // sleep for a while to see if we can recover
+                            // Sleep() will rethrow after waiting a bit
+                            // todo: make this smarter
+                            Sleep(tries,ex);
+                        }
+                        
+                        CurrentPageIndex = 0;
+
+                        if (CurrentPage.Length == 0)
+                        {
+                            // No listings were returned in the page
+                            // sleep for a while
+                            Sleep(tries);
+                        }
+                        else
+                        {
+                            tries = 0;
+                            break;
+                        }
+                    }
+                }
+                return true;
+            }
+
+            private void Sleep(int tries, Exception ex = null)
+            {
+                // wait up to 3 minutes between tries
+                int seconds = 180;
+
+                if (tries > 36)
+                    if (ex != null)
+                        throw ex;
+                else
+                    seconds = tries * 5;
+
+                System.Threading.Thread.Sleep(seconds*1000);
+            }
+
             public void Reset()
             {
                 After = Before = null;

RedditSharp/Things/Subreddit.cs

@@ -980,6 +980,47 @@ public Listing<ModAction> GetModerationLog(ModActionType action, string[] mods)
         {
             return new Listing<ModAction>(Reddit, string.Format(ModLogUrl + "?type={1}&mod={2}", Name, ModActionTypeConverter.GetRedditParamName(action), string.Join(",", mods)), WebAgent);
         }
+
+
+        /// <summary>
+        /// Infinitely yields new <see cref="Comment"/> posted to the subreddit.
+        /// </summary>
+        public IEnumerable<Comment> CommentStream
+        {
+            get
+            {
+                if (Name == "/")
+                    return new Listing<Comment>(Reddit, "/comments.json", WebAgent).GetListingStream();
+                return new Listing<Comment>(Reddit, string.Format(CommentsUrl, Name), WebAgent).GetListingStream();
+            }
+        }
+
+        /// <summary>
+        /// Infinitely yields new <see cref="Post"/> made to the subreddit.
+        /// </summary>
+        public IEnumerable<Post> SubmissionStream
+        {
+            get
+            {
+                if (Name == "/")
+                    return new Listing<Post>(Reddit, "/new.json", WebAgent).GetListingStream();
+                return new Listing<Post>(Reddit, string.Format(SubredditNewUrl, Name), WebAgent).GetListingStream();
+            }
+        }
+
+        /// <summary>
+        /// Infinitely yields new <see cref="ModAction"/> made on the subreddit.
+        /// </summary>
+        public IEnumerable<ModAction> ModerationLogStream
+        {
+            get
+            {
+                if (Name == "/")
+                    return new Listing<ModAction>(Reddit, string.Format(ModLogUrl, this.Name), WebAgent).GetListingStream();
+                return new Listing<ModAction>(Reddit, string.Format(ModLogUrl, this.Name), WebAgent).GetListingStream();
+            }
+        }
+
         #region Obsolete Getter Methods
 
         [Obsolete("Use Posts property instead")]