数据结构

Posted 2023-03-01 wpengch1

注：学习使用，禁止转载

数据结构概览

编写一个可维护的APP最难的方面之一是管理数据.在你的应用程序中,有很多种方式去获取数据.

• AJAX请求
• Websocket
• Indexdb
• LocalStorage
• Service Workers
• 等等

管理数据难的问题有:

• 我们怎么聚合这些不同来源的数据到一个一致的系统中
• 我们怎样才能避免意外的副作用引起的错误
• 我们怎么组织代码,使得对于其他参与者来说是清晰可见的
• 当数据变化的时候,怎么让我们的应用程序尽快做出反应

许多年来,MVC是一种标准的数据结构模式:Model包含领域模型逻辑,View负责显示数据,Controller将他们连接起来.

但是问题是,我们熟知的MVC模式不能直接很好的用在web开发中.

这里产生了一些新的数据结构的注意.

• MVW / Two-way data binding:Model-View-Whatever是angular1的默认结构.$scope提供了双向绑定,当数据变化的时候,整过应用程序会分享相同的数据.
• Flux,使用一个非双向数据结构.在Flux中,Stores保存数据,Views渲染数据,Actions在Stores中改变数据.这里有更多的方式设置Flux,但是这种方式是单向的,更容易理解.
• Observables,观察者模式.Observables给我们一个数据流,我们订阅数据流,然后执行操作去反应变化. 在javascript中,RxJs是最出名的反应式编程架构,它给了我们更多的能力,去组合操作进数据流.

angular2中的数据结构

对于数据结构的使用,angular2是非常灵活的.一个数据策略在这个项目用,不代表另外的项目就用这个数据策略.所以,angular2没有指定一个技术堆栈,而是可以使用在我们选用的任何数据结构之上.
这个的好处就是使得angular2可以使用在任何地方.不好的就是你必须自己选定一种模式.
别担心,我们不会让你自己来选择.在接下来的部分,我们会详细讲解使用这些模式怎么构建应用程序.

第一部分 Observables

Observables and RxJS

在angular中,我们可以使用Observables作为我们的数据结构,使用Observables组织我们的数据叫着Reactive Programming,反应式编程.

但是Observables是什么,Reactive Programming是怎么样的.Reactive Programming是一种跟异步数据流工作的方式.Observables是一种保存数据的方式,它可以使我们实现Reactive Programming.但是我怀疑,这样讲可能不清晰.所以,接下来让我们看看那例子.

注意:需要具备的知识

我想指出的是,这不是一本关于反应式编程的书,这里有一些反应式编程方面的书籍,我们将会在下面列出来.

考虑到这一章是一个将angular与RxJs结合的书,而不是详细介绍RxJs的书.

在这一章,我们会解释我们需要用到的RxJs的概念和API.但是记住,如果你对于RxJs还是新手,你需要阅读其他的治疗,以便去细致的了解.

学习反应式编程和RxJs

如果你刚开始学习RxJs,我建议你先读一下这篇文章.

• The introduction to Reactive Programming you’ve been missing46 by Andre Staltz

当你对RxJs有点了解的时候,下面的连接可以帮助你了解更多:

• Which static operators to use to create streams?
• Which instance operators to use on streams?
• RxMarbles - Interactive diagrams of the various operations on streams

在这章中,我们会提供RxJs的文档链接给你.RxJs的文档有很多的例子,解释了怎么组合不同的流和操作.

我会给你一个警告:开始学习RxJs的时候是比较难的.但是相信我,你可以掌握它,并且从中受益良多.

下面有一些关于RxJs的讨论可能对你有利:

• Promise发送单一的值,作为流发送多个值.
• 原始编程是拉数据,而反应式编程是推数据
• RxJs是函数式编程
• 流是可以组合的

聊天应用程序概览

在本章中,我们使用RxJs构建一个聊天应用程序.界面如下:

在这个例子中,我们会提供一些机器人去和你聊天.打开代码,并运行:

cd code/rxjs/chat
  npm install
  npm run go

然后打开浏览器输入:http://localhost:8080

注意关于这个应用程序的一些事情:

• 你可以点击threads可以切换聊天人
• 机器人会根据你的输入发送信息给你
• 顶部的未读消息数量是异步的

让我们看看这个应用程序的结构:

• 3个顶层angular 组件
• 3个model
• 3个service

Components 组件

我们可以将应用程序分解为3个组件

• ChatNavBar包含未读消息数
• ChatThreads聊天机器人列表
• ChatWindow展示与当前聊天对象的聊天信息

Models 模型

这个应用程序有三个model

• User存储聊天参与人的信息
• Message存储单条信息
• Thread存储聊天对象的聊天信息

Services 服务

在这个应用程序中,每一个model对应一个service,service是具有两类功能的单例:

• 提供我们应用程序可以订阅的数据流
• 提供增加或删除数据的操作

比如UserService:

• 发送当前用户的流
• 提供一个设置当前用户的方法

总结

在顶层,应用程序的数据结构是简单直接的.

• service维护发送模型的流
• components订阅流并根据最近的值渲染它们

比如,ChatThreads组件监听来自于ThreadService的最近thread列表.ChatWindow订阅最近消息列表.

在本章的其余部分,我们会深入分析怎样使用angular2 和RxJs来实现这个应用程序.

实现模型

User

code/rxjs/chat/app/ts/models.ts

export class User 
  id: string;

  constructor(public name: string,
              public avatarSrc: string) 
    this.id = uuid();
  

Thread

code/rxjs/chat/app/ts/models.ts

  id: string;
  lastMessage: Message;
  name: string;
  avatarSrc: string;

  constructor(id?: string,
              name?: string,
              avatarSrc?: string) 
    this.id = id || uuid();
    this.name = name;
    this.avatarSrc = avatarSrc;
  

注意,在我们的模型中,存储了lastMessage的引用,它让我们去显示最近的消息.

Message

export class Message 
  id: string;
  sentAt: Date;
  isRead: boolean;
  author: User;
  text: string;
  thread: Thread;

  constructor(obj?: any) 
    this.id              = obj && obj.id              || uuid();
    this.isRead          = obj && obj.isRead          || false;
    this.sentAt          = obj && obj.sentAt          || new Date();
    this.author          = obj && obj.author          || null;
    this.text            = obj && obj.text            || null;
    this.thread          = obj && obj.thread          || null;
  

上面要注意的是,我们的构造函数使用any类型,我们可以传递我们需要的参数进去,就可以新建一个Message,比如只传递id,或者isRead都可以.

实现UserService

UserService提供了一个存储当前用户的地方,并且当当前用户改变的时候,通知应用程序.

我们需要做的第一件事情是创建一个Typescript类,并使用@Injectable 注解该类,使其可注入.

code/rxjs/chat/app/ts/services/UserService.ts

@Injectable()
export class UserService 

可注入意思是可以将当前服务注入到其他的组件中.

currentUser

接下来,我们安装一个流,我们将会使用它去管理当前用户.

code/rxjs/chat/app/ts/services/UserService.ts

currentUser: Subject<User> = new BehaviorSubject<User>(null);

这里有一些事情需要解释:

• 我们定义了一个Subject流的实例变量
• 更具体的说,它是一个BehaviorSubject流,包含User对象
• 然而,这个流的第一个值是null

如果你没有使用过RxJs,你可能不知道Subject和BehaviorSubject是什么.你可以认为Subject是一个可读写流.

其结果之一是,因为消息会被立刻发送,一个新的订阅者可能会有丢失该消息的风险,BehaviorSubject可以解决这个问题.

BehaviorSubject有一个特定的属性来存储最近的值.这意味着任务订阅该流的人都会接收到最新的值,这个是很重要的,它意味着当参与者变化的时候,任何其他的订阅者都知道当前用户是谁.

设置一个新用户

当当前用户改变的时候,我们需要发布一个新的用户给流,我们可以使用两种方式来做这个事情.

直接增加一个新用户给流

最简单的方式就是直接新建一个用户给流,如下:

userService.subscribe((newUser) => 
  console.log('New User is: ', newUser.name);
)
// => New User is: originalUserName
let u = new User('Nate', 'anImgSrc'); userService.currentUser.next(u);
// => New User is: Nate

创建一个setCurrentUser(newUser: User)函数

另外一种方式就是可以创建一个新的函数setCurrentUser(newUser: User).如下:

code/rxjs/chat/app/ts/services/UserService.ts

public setCurrentUser(newUser: User): void 
    this.currentUser.next(newUser);

注意到了吗?我们始终使用currentUser的next方法,为什么还要做这一步呢?

这个是为了从流中解耦.通过使用currentUser包装方法使得其他的客户可以调用,而不会动我们里面的代码.

在这里,我不会直接建议你使用哪一种,但是对于一个大型应用程序来说,解耦是很重要的.

UserService.ts

UserService像下面这样.

code/rxjs/chat/app/ts/services/UserService.ts

import Injectable, bind from '@angular/core';
import Subject, BehaviorSubject from 'rxjs';
import User from '../models';


/**
 * UserService manages our current user
 */
@Injectable()
export class UserService 
  // `currentUser` contains the current user
  currentUser: Subject<User> = new BehaviorSubject<User>(null);

  public setCurrentUser(newUser: User): void 
    this.currentUser.next(newUser);
  


export var userServiceInjectables: Array<any> = [
  bind(UserService).toClass(UserService)
];

MessagesService

在这个应用程序中,所有的消息都是通过MessagesService管理.

比起UserService来说,MessageService有更多的流,这里有5个流,3个数据管理流和2个行为流

三个数据管理流是:

• newMessages 发送每个新消息一次
• messages 发送当前消息列表
• updates 在messages上执行操作

newMessages流

newMessages流仅仅发送一次新消息

code/rxjs/chat/app/ts/services/MessagesService.ts

@Injectable()
export class MessagesService 
// a stream that publishes new messages only once newMessages: Subject<Message> = new Subject<Message>();

如果你希望,你可以定义一个帮助函数,实现添加新消息.

code/rxjs/chat/app/ts/services/MessagesService.ts

addMessage(message: Message): void 
    this.newMessages.next(message);

它可能是有帮助的,有一个流从thread中所有消息,而不是参与用户.比如,考虑Echo Bot:

当我们实现Echo Bot的时候,不希望有一个无限循环,或者重复以前的信息.

为了实现这个功能,我们可以订阅newMessage流,并且过滤掉所有信息,具有这些条件的:

• 这个thread的消息
• 不是被机器人输入的

你可以认为这个就是说我希望我与某个机器人的聊天信息,除了机器人说的之外:
code/rxjs/chat/app/ts/services/MessagesService.ts

messagesForThreadUser(thread: Thread, user: User): Observable<Message> 
    return this.newMessages
      .filter((message: Message) => 
               // belongs to this thread
        return (message.thread.id === thread.id) &&
               // and isn't authored by this user
               (message.author.id !== user.id);
); 

messages流

当newMessages发送一个新消息的时候,messages发送最近消息的数组

code/rxjs/chat/app/ts/services/MessagesService.ts

// `messages` is a stream that emits an array of the most up to date messages messages: Observable<Message[]>;

考虑messages怎么获取,在讨论之前,我们需要先讨论updates流和一个新的模式:操作流

操作流

这里:

• 我们会保存存储最近消息列表的消息流(messages)状态
• 我们会使用一个update流应用到messages流上

你可以使用这种方式理解:放入update流上的任何函数,都会改变当前消息流列表.这个函数应该是接收一个消息列表,并且返回一个新的消息列表.让我们看看接口代码:

code/rxjs/chat/app/ts/services/MessagesService.ts

interface IMessagesOperation extends Function 
  (messages: Message[]): Message[];

让我们定义我们的update流

code/rxjs/chat/app/ts/services/MessagesService.ts

// `updates` receives _operations_ to be applied to our `messages`
// it's a way we can perform changes on *all* messages (that are currently 
// stored in `messages`)
updates: Subject<any> = new Subject<any>();

记住,updates接收一个应用于我们消息列表的操作.但是我们怎么连接起来?我们可以像这样:

code/rxjs/chat/app/ts/services/MessagesService.ts

constructor() 
    this.messages = this.updates
      // watch the updates and accumulate operations on the messages
      .scan((messages: Message[],
             operation: IMessagesOperation) => 
               return operation(messages);
             ,
            initialMessages)

这个代码介绍了一个新的流函数:scan.如果你对函数式编程很熟悉,那么它跟reduce很像.它对数组中的每一个元素运行一次这个函数,并最后累积起来.scan有什么特殊的呢?它会立刻发送每一个元素的结果,而不会等待所有操作完成,这个正式我们需要的.

当我们调用this.updates.scan时,我们会创建一个新的流,它被updates订阅,在每一个部分,我们会获得:

• 我们已经完成的messages
• new操作会被应用

最后,我们返回new message[]

共享一个流

关于流的一个信息就是它们默认是不共享的.意思就是,如果一个订阅者从流中读取了一个值,这个值就会永远消失.在我们的例子中,我们希望,1. 在许多订阅者中共享流. 2. 对于来迟的订阅者,发送最近的流.

为了做这个事情,我们有两个操作:publishReplay和refCount.

• publishReplay让我们可以在多个订阅者中共享流,并且对于将来的订阅者返回一个消息数量n
• refCount使得发送的数量容易被使用

code/rxjs/chat/app/ts/services/MessagesService.ts

// watch the updates and accumulate operations on the messages
      .scan((messages: Message[],
             operation: IMessagesOperation) => 
               return operation(messages);
             ,
            initialMessages)
      // make sure we can share the most recent list of messages across anyone
      // who's interested in subscribing and cache the last known list of
      // messages
      .publishReplay(1)
      .refCount();

向messages流中增加message

我们可以像下面这样增加:

var myMessage = new Message(/* params here... */);
updates.next( (messages: Message[]): Message[] => 
  return messages.concat(myMessage);
)

上面,我们给updates流增加了一个操作.messages订阅了该流,所以当增加的时候,它会将其添加到自己的消息列表中去.

上面的方法有一个问题,我们每次都会去调用这个方法.其实我们可以像下面这样做:

Data Architecture with Observables - Part 1: Services 170
 addMessage(newMessage: Message) 
  updates.next( (messages: Message[]): Message[] => 
    return messages.concat(newMessage);
  )

使用下面的调用就可以了:

var myMessage = new Message(/* params here... */);
MessagesService.addMessage(myMessage);

方法很好,但是它不是反应式的.部分的,由于创建消息的行为不能与其他流组合(同时这个方法也绕过了NewMessage流).

创建一个新消息的反应式方式可能是有一个流,它接收一个消息列表并且增加到这个列表中去.
又一次,如果你没有这样思考的话,这个方式有点新.下面是你怎么实现它:

• 我们创建一个叫create的action stream-行为流(action stream只是用来区别它的功能,本质上它还是一个Subject流)
  code/rxjs/chat/app/ts/services/MessagesService.ts

// action streams
create: Subject<Message> = new Subject<Message>();

• 接下来,在我们的构造器中配置create流
  code/rxjs/chat/app/ts/services/MessagesService.ts

this.create
      .map( function(message: Message): IMessagesOperation 
        return (messages: Message[]) => 
          return messages.concat(message);
; 
)

这个map方法有点像JavaScript数组的map方法,它针对每一个元素运行这个函数并返回函数返回值.

这个流会发送一个接收一个消息列表并且添加这个消息到消息列表的方法.

现在,我们有了create流,我们需要将其挂到updates流上去,通过subscribe完成这个功能.

code/rxjs/chat/app/ts/services/MessagesService.ts

this.create
      .map( function(message: Message): IMessagesOperation 
        return (messages: Message[]) => 
          return messages.concat(message);
; )
      .subscribe(this.updates);

上面的意思是我们updates流订阅去监听create流.这个意思就是说,如果create流接收到一个消息,它会发送将会被updates流接收的一个IMessagesOperation操作然后message将会被增加到messages里面去.

下面的图描述了这个场景:

好主意,它意味着我们可以获得下面的东西:

• 来自于messages的消息列表
• 在当前列表上面执行操作updates的方式
• 一个易于使用的create流,它可以添加到我们的updates流上去.

在这个代码中,如果你想获取最近的消息列表,你仅仅需要去获取messages流.

但是有一个问题,我们始终没有将这个流连接到newMessages流上.

要是有一种方便的方法做这个事情就好了,事实证明是有的,如下:

code/rxjs/chat/app/ts/services/MessagesService.ts

this.newMessages
      .subscribe(this.create);

现在,场景如下:

现在,我们的工作流完成了.这是一个两全其美的办法:如果我们需要订阅单个信息,可以通过newMessages流,如果我们想要最近的消息列表,我们可以订阅messages流.

MessagesService全部.

import Injectable, bind from '@angular/core';
import Subject, Observable from 'rxjs';
import User, Thread, Message from '../models';

let initialMessages: Message[] = [];

interface IMessagesOperation extends Function 
  (messages: Message[]): Message[];


@Injectable()
export class MessagesService 
  // a stream that publishes new messages only once
  newMessages: Subject<Message> = new Subject<Message>();

  // `messages` is a stream that emits an array of the most up to date messages
  messages: Observable<Message[]>;

  // `updates` receives _operations_ to be applied to our `messages`
  // it's a way we can perform changes on *all* messages (that are currently 
  // stored in `messages`)
  updates: Subject<any> = new Subject<any>();

  // action streams
  create: Subject<Message> = new Subject<Message>();
  markThreadAsRead: Subject<any> = new Subject<any>();

  constructor() 
    this.messages = this.updates
      // watch the updates and accumulate operations on the messages
      .scan((messages: Message[],
             operation: IMessagesOperation) => 
               return operation(messages);
             ,
            initialMessages)
      // make sure we can share the most recent list of messages across anyone
      // who's interested in subscribing and cache the last known list of
      // messages
      .publishReplay(1)
      .refCount();

    // `create` takes a Message and then puts an operation (the inner function)
    // on the `updates` stream to add the Message to the list of messages.
    //
    // That is, for each item that gets added to `create` (by using `next`)
    // this stream emits a concat operation function.
    //
    // Next we subscribe `this.updates` to listen to this stream, which means
    // that it will receive each operation that is created
    //
    // Note that it would be perfectly acceptable to simply modify the
    // "addMessage" function below to simply add the inner operation function to
    // the update stream directly and get rid of this extra action stream
    // entirely. The pros are that it is potentially clearer. The cons are that
    // the stream is no longer composable.
    this.create
      .map( function(message: Message): IMessagesOperation 
        return (messages: Message[]) => 
          return messages.concat(message);
        ;
      )
      .subscribe(this.updates);

    this.newMessages
      .subscribe(this.create);

    // similarly, `markThreadAsRead` takes a Thread and then puts an operation
    // on the `updates` stream to mark the Messages as read
    this.markThreadAsRead
      .map( (thread: Thread) => 
        return (messages: Message[]) => 
          return messages.map( (message: Message) => 
            // note that we're manipulating `message` directly here. Mutability
            // can be confusing and there are lots of reasons why you might want
            // to, say, copy the Message object or some other 'immutable' here
            if (message.thread.id === thread.id) 
              message.isRead = true;
            
            return message;
          );
        ;
      )
      .subscribe(this.updates);

  

  // an imperative function call to this action stream
  addMessage(message: Message): void 
    this.newMessages.next(message);
  

  messagesForThreadUser(thread: Thread, user: User): Observable<Message> 
    return this.newMessages
      .filter((message: Message) => 
               // belongs to this thread
        return (message.thread.id === thread.id) &&
               // and isn't authored by this user
               (message.author.id !== user.id);
      );
  


export var messagesServiceInjectables: Array<any> = [
  bind(MessagesService).toClass(MessagesService)
];

试试

如果还没有试,这里我们提供了一个方式去感受MessagesService是怎么工作的.测试在test/services/MessagesService.spec.ts.

:fa-info-circle:为了运行测试,打开终端,运行下列代码:

cd /path/to/code/rxjs/chat//<--yourpathwillvary 
npm install
karma start

首先,让我们创建一些要使用的模型例子:

code/rxjs/chat/test/services/MessagesService.spec.ts

import Message, User, Thread from '../../app/ts/models'; describe('MessagesService', () => 
  it('should test', () => 
    let user: User = new User('Nate', '');
    let thread: Thread = new Thread('t1', 'Nate', '');
    let m1: Message = new Message(
      author: user,
      text: 'Hi!',
      thread: thread
);

接下来,让我们订阅流:
code/rxjs/chat/test/services/MessagesService.spec.ts

text: 'Bye!',
      thread: thread
    );

    let messagesService: MessagesService = new MessagesService();

    // listen to each message indivdually as it comes in
    messagesService.newMessages
      .subscribe( (message: Message) => 
        console.log('=> newMessages: ' + message.text);
      );

    // listen to the stream of most current messages
    messagesService.messages
      .subscribe( (messages: Message[]) => 
        console.log('=> messages: ' + messages.length);
      );

    messagesService.addMessage(m1);
    messagesService.addMessage(m2);

注意,虽然我们首先订阅newMessages,并且newMessages被addMessage直接调用,但是我们的messages首先打印出信息.这是因为Messages订阅newMessages早于我们测试中的初始化(当MessageService被初始化时).你的代码不能依赖于这个顺序,但是了解它是有必要的.

ThreadsService

ThreadsService中,我们会定义各自发送的四个流:

• 当前线程的map
• 当前线程的有序列表,newnewest-firt(时间最新的在最前)
• 当前选择线程
• 当前选择线程的消息列表

线程的map

code/rxjs/chat/app/ts/services/ThreadsService.ts

import Injectable, bind from '@angular/core';
import Subject, BehaviorSubject, Observable from 'rxjs';
import Thread, Message from '../models';
import MessagesService from './MessagesService';
import * as _ from 'underscore';

@Injectable()
export class ThreadsService 

  // `threads` is a observable that contains the most up to date list of threads
  threads: Observable< [key: string]: Thread >;

注意这个流将会发送一个map,key为线程id,value为线程本身.

为了保存当前线程列表,我们创建一个threads流关联MessageService.messages流.

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.threads = messagesService.messages

每一次addMessage增加时,messages会发送一个数组,我们将会检查每个message,并且返回我们想要的thread列表.

注意,上面我们每次发送一个新的列表.因为可能下一次可能删除一些消息(比如离开会话).因为我们每次都会重新计算thread的列表,正常情况下,如果没有消息,我们会删除thread列表.

在线程列表中,我们会显示最近聊天的thread.

如下:

为了做这个事情,我们会存储每个thread的最近消息列表,通过比较sentAt时间获取最新的thread

code/rxjs/chat/app/ts/services/ThreadsService.ts

// Cache the most recent message for each thread
let messagesThread: Thread = threads[message.thread.id]; 
if (!messagesThread.lastMessage ||
messagesThread.lastMessage.sentAt < message.sentAt)  
    messagesThread.lastMessage = message;

);
return threads; );

结合起来,threads像下面这样:

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.threads = messagesService.messages
      .map( (messages: Message[]) => 
        let threads: [key: string]: Thread = ;
        // Store the message's thread in our accumulator `threads`
        messages.map((message: Message) => 
          threads[message.thread.id] = threads[message.thread.id] ||
            message.thread;

          // Cache the most recent message for each thread
          let messagesThread: Thread = threads[message.thread.id];
          if (!messagesThread.lastMessage ||
              messagesThread.lastMessage.sentAt < message.sentAt) 
            messagesThread.lastMessage = message;
          
        );
        return threads;
      );

实施ThreadsService

code/rxjs/chat/test/services/ThreadsService.spec.ts

import 
  it,
  describe
 from '@angular/core/testing';

import MessagesService, ThreadsService from '../../app/ts/services/services';
import Message, User, Thread from '../../app/ts/models';
import * as _ from 'underscore';

describe('ThreadsService', () => 
  it('should collect the Threads from Messages', () => 

    let nate: User = new User('Nate Murray', '');
    let felipe: User = new User('Felipe Coury', '');

    let t1: Thread = new Thread('t1', 'Thread 1', '');
    let t2: Thread = new Thread('t2', 'Thread 2', '');

    let m1: Message = new Message(
      author: nate,
      text: 'Hi!',
      thread: t1
    );

    let m2: Message = new Message(
      author: felipe,
      text: 'Where did you get that hat?',
      thread: t1
    );

    let m3: Message = new Message(
      author: nate,
      text: 'Did you bring the briefcase?',
      thread: t2
    );

    let messagesService: MessagesService = new MessagesService();
    let threadsService: ThreadsService = new ThreadsService(messagesService);

    threadsService.threads
      .subscribe( (threadIdx:  [key: string]: Thread ) => 
        let threads: Thread[] = _.values(threadIdx);
        let threadNames: string = _.map(threads, (t: Thread) => t.name)
                                   .join(', ');
        console.log(`=> threads ($threads.length): $threadNames `);
      );

    messagesService.addMessage(m1);
    messagesService.addMessage(m2);
    messagesService.addMessage(m3);

    // => threads (1): Thread 1
    // => threads (1): Thread 1
    // => threads (2): Thread 1, Thread 2

  );
);

thread的有序列表,最新消息在前面

让我们创建一个新的thream.它返回有序的thread列表,基于时间.orderedThreads.
code/rxjs/chat/app/ts/services/ThreadsService.ts

// `orderedThreads` contains a newest-first chronological list of threads orderedThreads: Observable<Thread[]>;

接下来,在构造器中,我们定义了orderedThreads,它订阅了threads,并且返回有序列表.

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.orderedThreads = this.threads .map((threadGroups:  [key: string]: Thread ) => 
let threads: Thread[] = _.values(threadGroups);
        return _.sortBy(threads, (t: Thread) => t.lastMessage.sentAt).reverse();
      );

当前选择的thread – currentThread

我们的应用程序需要知道当前选择的是thread,这个让我们知道:

• 在消息窗口中应该显示那个thread
• 在thread列表中,应该标识那个thread为当前thread

让我们创建一个存储当前thread的BehaviorSubject流 – currentThread

code/rxjs/chat/app/ts/services/ThreadsService.ts

// `currentThread` contains the currently selected thread
  currentThread: Subject<Thread> =
new BehaviorSubject<Thread>(new Thread());

首先,我们分配一个空的thread作为默认值.这里就不需要进一步的配置设置了.

设置当前thread

为了设置当前thread,我们可以

• 通过next直接提交一个新的thread
• 给它增加一个新的辅助方法

让我们定义一个帮助函数,setCurrentThread,我们可以使用它设置thread.

code/rxjs/chat/app/ts/services/ThreadsService.ts

setCurrentThread(newThread: Thread): void 
    this.currentThread.next(newThread);

将当前thread标记为已读

我们想要去跟踪未读消息的数量，当我们切换到一个thread的时候，我们希望标记该thread的所有消息为已读。我们需要去做下面这些内容：

1. messagesService.makeThreadAsRead接收一个thread，然后将该thread对应的所有消息标记为已读
2. 我们的currentThread代表当前thread发送一个单一的thread

所以我们需要做的就是将他们组合在一起。

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.currentThread.subscribe(this.messagesService.markThreadAsRead);

当前thread（currentThread）的消息列表（currentThreadMessages）

既然我们已经有选择了当前thread，我们要确保我们能够显示该thread的消息列表。

实现这个效果比表面看起来要复杂得多，我们像下面这样去实现：

var theCurrentThread:Thread;

this.currentThread.subscribe((thread:Thread) => 
    theCurrentThread = thread;
);

this.currentThreadMessages.map(
    (mesages:Message[]) => 
        return _.filter(messages,
            (message:Message) => 
                return message.thread.id == theCurrentThread.id;
         )
 );

这种方法有什么问题呢？如果currentThread改变了，currentThreadMessages不知道，因此，我们会得到一个过时的curentMessages。

现在如果我们改变它，将消息列表存储为一个变量，让该变量去订阅currentThread的变化，但是现在又有另一个问题，我们仅仅知道currentThread变化了，但是当有新的消息进来的时，我们并不知道。

怎么解决这个问题呢？

事实证明，RxJs有一组操作，我们可以使用它们来组合多个流。在这个例子中，我们希望“不管是currentThreadmessageService.messages改变，我们都希望去发射一些东西”，我们使用combineLatest操作符来做这个事情。

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.currentThreadMessages = this.currentThread
 .combineLatest(messagesService.messages,
 (currentThread: Thread, messages: Message[]) => 

当我们结合两个流的时候，那个先到是不保证的。所以我们需要去检查是否有我们需要的内容，不然我们只能得到一个空的列表。

现在，我们有了currentThread和messages，我们只需要去找到我们感兴趣的消息就可以了。

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.currentThreadMessages = this.currentThread.combineLatest(messagesService.messages,
     (currentThread: Thread, messages: Message[]) => 
         if (currentThread && messages.length > 0) 
             return _.chain(messages).filter((message: Message) =>(message.thread.id === currentThread.id))

另一个细节，因为我们已经知道了当前thread的消息，所以将当前thread的消息标记为已读就很方便了。

code/rxjs/chat/app/ts/services/ThreadsService.ts

return _.chain(messages)
 .filter((message: Message) =>
 (message.thread.id === currentThread.id)) .map((message: Message) => 
     message.isRead = true;
 return message; ) .value();

将所有的东西组合起来，我们的currentThreadMessages像下面这样：

code/rxjs/chat/app/ts/services/ThreadsService.ts

this.currentThreadMessages = this.currentThread
      .combineLatest(messagesService.messages,
                     (currentThread: Thread, messages: Message[]) => 
        if (currentThread && messages.length > 0) 
          return _.chain(messages)
            .filter((message: Message) =>
                    (message.thread.id === currentThread.id))
            .map((message: Message) => 
              message.isRead = true;
              return message; )
            .value();
         else 
          return [];
        
      );

完整的ThreadsService

import Injectable, bind from '@angular/core';
import Subject, BehaviorSubject, Observable from 'rxjs';
import Thread, Message from '../models';
import MessagesService from './MessagesService';
import * as _ from 'underscore';

@Injectable()
export class ThreadsService 

  // `threads` is a observable that contains the most up to date list of threads
  threads: Observable< [key: string]: Thread >;

  // `orderedThreads` contains a newest-first chronological list of threads
  orderedThreads: Observable<Thread[]>;

  // `currentThread` contains the currently selected thread
  currentThread: Subject<Thread> =
    new BehaviorSubject<Thread>(new Thread());

  // `currentThreadMessages` contains the set of messages for the currently
  // selected thread
  currentThreadMessages: Observable<Message[]>;

  constructor(public messagesService: MessagesService) 

    this.threads = messagesService.messages
      .map( (messages: Message[]) => 
        let threads: [key: string]: Thread = ;
        // Store the message's thread in our accumulator `threads`
        messages.map((message: Message) => 
          threads[message.thread.id] = threads[message.thread.id] ||
            message.thread;

          // Cache the most recent message for each thread
          let messagesThread: Thread = threads[message.thread.id];
          if (!messagesThread.lastMessage ||
              messagesThread.lastMessage.sentAt < message.sentAt) 
            messagesThread.lastMessage = message;
          
        );
        return threads;
      );

    this.orderedThreads = this.threads
      .map((threadGroups:  [key: string]: Thread ) => 
        let threads: Thread[] = _.values(threadGroups);
        return _.sortBy(threads, (t: Thread) => t.lastMessage.sentAt).reverse();
      );

    this.currentThreadMessages = this.currentThread
      .combineLatest(messagesService.messages,
                     (currentThread: Thread, messages: Message[]) => 
        if (currentThread && messages.length > 0) 
          return _.chain(messages)
            .filter((message: Message) =>
                    (message.thread.id === currentThread.id))
            .map((message: Message) => 
              message.isRead = true;
              return message; )
            .value();
         else 
          return [];
        
      );

    this.currentThread.subscribe(this.messagesService.markThreadAsRead);
  

  setCurrentThread(newThread: Thread): void 
    this.currentThread.next(newThread);
  



export var threadsServiceInjectables: Array<any> = [
  bind(ThreadsService).toClass(ThreadsService)
];

数据模型总结

我们的数据模型和服务，我们已经具备了编写组件的一切基础。在下一章，我们会编写三个主要的组件。