Adding a recurring inbox fetch.

- Fixes #94

ui/src/components/main.tsx

@@ -2,7 +2,7 @@ import { Component } from 'inferno';
 import { Link } from 'inferno-router';
 import { Subscription } from "rxjs";
 import { retryWhen, delay, take } from 'rxjs/operators';
-import { UserOperation, CommunityUser, GetFollowedCommunitiesResponse, ListCommunitiesForm, ListCommunitiesResponse, Community, SortType, GetSiteResponse, GetRepliesResponse, GetRepliesForm, ListingType } from '../interfaces';
+import { UserOperation, CommunityUser, GetFollowedCommunitiesResponse, ListCommunitiesForm, ListCommunitiesResponse, Community, SortType, GetSiteResponse, ListingType } from '../interfaces';
 import { WebSocketService, UserService } from '../services';
 import { PostListings } from './post-listings';
 import { msgOp, repoUrl, mdToHtml } from '../utils';
@@ -60,15 +60,6 @@ export class Main extends Component<MainProps, MainState> {
 
     if (UserService.Instance.user) {
       WebSocketService.Instance.getFollowedCommunities();
-
-      // Get replies for the count
-      let repliesForm: GetRepliesForm = {
-        sort: SortType[SortType.New],
-        unread_only: true,
-        page: 1,
-        limit: 9999,
-      };
-      WebSocketService.Instance.getReplies(repliesForm);
     }
 
     let listCommunitiesForm: ListCommunitiesForm = {
@@ -190,14 +181,7 @@ export class Main extends Component<MainProps, MainState> {
       this.state.site.site = res.site;
       this.state.site.banned = res.banned;
       this.setState(this.state);
-    } else if (op == UserOperation.GetReplies) {
-      let res: GetRepliesResponse = msg;
-      this.sendRepliesCount(res);
     } 
   }
-
-  sendRepliesCount(res: GetRepliesResponse) {
-    UserService.Instance.sub.next({user: UserService.Instance.user, unreadCount: res.replies.filter(r => !r.read).length});
-  }
 }
 

ui/src/components/navbar.tsx

@@ -1,6 +1,10 @@
 import { Component, linkEvent } from 'inferno';
 import { Link } from 'inferno-router';
-import { UserService } from '../services';
+import { Subscription } from "rxjs";
+import { retryWhen, delay, take } from 'rxjs/operators';
+import { WebSocketService, UserService } from '../services';
+import { UserOperation, GetRepliesForm, GetRepliesResponse, SortType } from '../interfaces';
+import { msgOp } from '../utils';
 import { version } from '../version';
 
 interface NavbarState {
@@ -11,7 +15,8 @@ interface NavbarState {
 }
 
 export class Navbar extends Component<any, NavbarState> {
-
+  private wsSub: Subscription;
+  private userSub: Subscription;
   emptyState: NavbarState = {
     isLoggedIn: (UserService.Instance.user !== undefined),
     unreadCount: 0,
@@ -24,12 +29,22 @@ export class Navbar extends Component<any, NavbarState> {
     this.state = this.emptyState;
     this.handleOverviewClick = this.handleOverviewClick.bind(this);
 
+    this.keepFetchingReplies();
+
     // Subscribe to user changes
-    UserService.Instance.sub.subscribe(user => {
+    this.userSub = UserService.Instance.sub.subscribe(user => {
       this.state.isLoggedIn = user.user !== undefined;
       this.state.unreadCount = user.unreadCount;
       this.setState(this.state);
     });
+
+    this.wsSub = WebSocketService.Instance.subject
+    .pipe(retryWhen(errors => errors.pipe(delay(3000), take(10))))
+    .subscribe(
+      (msg) => this.parseMessage(msg),
+        (err) => console.error(err),
+        () => console.log('complete')
+    );
   }
 
   render() {
@@ -38,6 +53,11 @@ export class Navbar extends Component<any, NavbarState> {
     )
   }
 
+  componentWillUnmount() {
+    this.wsSub.unsubscribe();
+    this.userSub.unsubscribe();
+  }
+
   // TODO class active corresponding to current page
   // TODO toggle css collapse
   navbar() {
@@ -117,5 +137,37 @@ export class Navbar extends Component<any, NavbarState> {
     i.state.expanded = !i.state.expanded;
     i.setState(i.state);
   }
+
+  parseMessage(msg: any) {
+    let op: UserOperation = msgOp(msg);
+    if (msg.error) {
+      alert(msg.error);
+      return;
+    } else if (op == UserOperation.GetReplies) {
+      let res: GetRepliesResponse = msg;
+      this.sendRepliesCount(res);
+    } 
+  }
+
+  keepFetchingReplies() {
+    this.fetchReplies();
+    setInterval(() => this.fetchReplies(), 30000);
+  }
+
+  fetchReplies() {
+    if (this.state.isLoggedIn) {
+      let repliesForm: GetRepliesForm = {
+        sort: SortType[SortType.New],
+        unread_only: true,
+        page: 1,
+        limit: 9999,
+      };
+      WebSocketService.Instance.getReplies(repliesForm);
+    }
+  }
+
+  sendRepliesCount(res: GetRepliesResponse) {
+    UserService.Instance.sub.next({user: UserService.Instance.user, unreadCount: res.replies.filter(r => !r.read).length});
+  }
 }