Thread unread count (Issue#49)

src/app/chat-thread/chat-thread.component.html

@@ -6,6 +6,7 @@
   <div class="media-body">
     <h5 class="media-heading contact-name">{{thread.name}}
       <span *ngIf="selected">&bull;</span>
+      <span class="badge">{{ unread?.sum }}</span>
     </h5>
     <small class="message-preview">{{thread.lastMessage.text}}</small>
   </div>

src/app/chat-thread/chat-thread.component.ts

@@ -16,6 +16,7 @@ import { Thread } from '../thread/thread.model';
 })
 export class ChatThreadComponent implements OnInit {
   @Input() thread: Thread;
+  @Input() unread: any;
   selected = false;
 
   constructor(public threadsService: ThreadsService) {

src/app/chat-threads/chat-threads.component.html

@@ -4,7 +4,7 @@
 
     <chat-thread
          *ngFor="let thread of threads | async"
-         [thread]="thread">
+         [thread]="thread" [unread]="unreadThread[thread.id]">
     </chat-thread>
 
   </div>

src/app/chat-threads/chat-threads.component.ts

@@ -3,19 +3,53 @@ import {
   OnInit,
   Inject
 } from '@angular/core';
+import * as _ from 'lodash';
 import { Observable } from 'rxjs';
 import { Thread } from '../thread/thread.model';
+import { Message } from '../message/message.model';
 import { ThreadsService } from './../thread/threads.service';
+import { MessagesService } from './../message/messages.service';
 
 @Component({
   selector: 'chat-threads',
   templateUrl: './chat-threads.component.html',
   styleUrls: ['./chat-threads.component.css']
 })
-export class ChatThreadsComponent {
+export class ChatThreadsComponent implements OnInit {
   threads: Observable<any>;
+  unreadThread: any;
 
-  constructor(public threadsService: ThreadsService) {
+  constructor(public threadsService: ThreadsService, public messagesService: MessagesService) {
     this.threads = threadsService.orderedThreads;
   }
+
+  ngOnInit() {
+    this.messagesService.messages
+      .combineLatest(
+        this.threadsService.currentThread,
+        (messages: Message[], currentThread: Thread) =>
+          [currentThread, messages] )
+
+      .subscribe(([currentThread, messages]: [Thread, Message[]]) => {
+        this.unreadThread = {};
+        _.reduce(
+          messages,
+          (sum: number, m: Message) => {
+            const messageIsInCurrentThread: boolean = m.thread &&
+              currentThread &&
+              (currentThread.id === m.thread.id);
+                if (m && !m.isRead && !messageIsInCurrentThread) {
+                  sum = sum + 1;
+                  let cid = m.thread.id;
+                  if (this.unreadThread[cid]) {
+                    this.unreadThread[cid].sum = this.unreadThread[cid].sum + 1;
+                  } else {
+                    this.unreadThread[cid] = { sum : 1 };
+                  }
+                }
+                return sum;
+          },
+        0);
+      });
+  }
 }

src/app/message/messages.service.ts

@@ -4,6 +4,8 @@ import { User } from '../user/user.model';
 import { Thread } from '../thread/thread.model';
 import { Message } from '../message/message.model';
 
+import * as _ from 'lodash';
+
 const initialMessages: Message[] = [];
 
 interface IMessagesOperation extends Function {
@@ -26,6 +28,8 @@ export class MessagesService {
   // action streams
   create: Subject<Message> = new Subject<Message>();
   markThreadAsRead: Subject<any> = new Subject<any>();
+  remove: Subject<Message> = new Subject<Message>();
+  removeThreadMessages: Subject<any> = new Subject<any>();
 
   constructor() {
     this.messages = this.updates
@@ -84,13 +88,41 @@ export class MessagesService {
       })
       .subscribe(this.updates);
 
+    // Remove all messages from particular thread
+    this.removeThreadMessages
+      .map( (thread: Thread) => {
+        return (messages: any) => {
+          _.remove(messages, (el: any) => {
+            return el.id === thread.id
+          });
+          return messages;
+        };
+      })
+      .subscribe(this.updates);
+
+    // Remove all messages from message stream
+    this.remove
+      .map( function(): IMessagesOperation {
+        return (messages: Message[]) => {
+          return messages = [];
+        };
+      })
+      .subscribe(this.updates);
   }
 
   // an imperative function call to this action stream
   addMessage(message: Message): void {
     this.newMessages.next(message);
   }
 
+  removeAllMessages() {
+    this.remove.next();
+  }
+
+  removeAllThreadMessages(thread: Thread) {
+    this.removeThreadMessages.next(thread);
+  }
+
   messagesForThreadUser(thread: Thread, user: User): Observable<Message> {
     return this.newMessages
       .filter((message: Message) => {