软件简介

• 代码精简、可读性强（尽可能用最少的代码实现）
• 设计精妙、实现优雅（尽可能利用golang的语言特点和优势）
• 可扩展性强（可自定义Observable以及Operator）
• 占用系统资源低（尽一切可能减少创建goroutine和其他对象）
• 性能强（尽一切可能减少计算量）

每一行代码都是深思熟虑……

已实现的功能

Observable

FromSlice FromChan Of Range Subject
Timeout Interval Merge Concat Race
CombineLatest Empty Never Throw

Operator

Do Take TakeWhile TakeUntil Skip SkipWhile
SkipUntil IgnoreElements Share StartWith Zip
Filter Distinct DistinctUntilChanged Debounce
DebounceTime Throttle ThrottleTime First Last
Count Max Min Reduce Map MapTo
MergeMap MergeMapTo SwitchMap SwitchMapTo

使用方法

链式调用方式

import (
    . "github.com/langhuihui/RxGo/rx"
)
func main(){
    err := Of(1, 2, 3, 4).Take(2).Subscribe(ObserverFunc(func(event *Event) {

    }))
}

管道模式

import (
    . "github.com/langhuihui/RxGo/rx"
    . "github.com/langhuihui/RxGo/pipe"
)
func main(){
    err := Of(1, 2, 3, 4).Pipe(Skip(1),Take(2)).Subscribe(ObserverFunc(func(event *Event) {

    }))
}

管道模式相比链式模式，具有操作符 可扩展性 ，用户可以按照规则创建属于自己的操作符

type Operator func(Observable) Observable

操作符只需要返回Operator这个类型即可,例如 实现一个happy为false就立即完成的操作符

func MyOperator(happy bool) Operator {
    return func(source Observable) Observable {
        return func (sink *Observer) error {
            if happy{
                return source(sink)
            }
            return nil
        }
    }
}

创建自定义Observable

在任何时候，您都可以创建自定义的Observable，用来发送任何事件

import (
    . "github.com/langhuihui/RxGo/rx"
)
func MyObservable (sink *Control) error {
    sink.Next("hello")
    return nil
}
func main(){
    ob := Observable(MyObservable)
    ob.Subscribe(ObserverFunc(func(event *Event) {

    }))
}

设计思想

基本知识

所谓Observable，就是一个可以被订阅，然后不断发送事件的事件源，见如下示意图

                                time -->

(*)-------------(o)--------------(o)---------------(x)----------------|>
 |               |                |                 |                 |
Start          value            value             error              Done

该示意图代表了，事件被订阅后（Start）开始不停发送事件的过程，直到发出error或者Done（完成）为止

有的Observable并不会发出完成事件，比如Never

参考网站： rxmarbles

总体方案

实现Rx的关键要素，是要问几个问题

1. 如何定义Observable，？（一个结构体？一个函数？一个接口？一个Channel？）
2. 如何实现 订阅 逻辑？（调用函数？发送数据？）
3. 如何实现 接受数据 ？（如何实现Observer？）
4. 如何实现 完成/错误 的传递？
5. 如何实现 取消订阅 ？（难点：在事件响应中取消，以及在任何其他goroutine中取消订阅）
6. 如何实现 操作符 ？
7. 操作符 如何处理连锁反应，比如后面描述的情况
8. 如何实现链式和管道两种编程模式
9. 如何让用户扩展（自定义）Observable和 操作符

10. 如何向普通用户解释复杂的概念

11. 当用户需要订阅或者终止事件流，则进行链路传递，订阅或者终止所有中间过程中的事件源

    Observable---------Operator----------Operator-----------Observer
    <| <| <|
    订阅/取消 订阅/取消 订阅/取消

12. 当事件流完成或者报错时，需要通知下游事件流的完成或者报错

    Observable---------Operator----------Operator-----------Observer
    |> |> |>
    完成/错误 完成/错误 完成/错误

实际情况远比这个复杂，后面会进行分析

可观察对象（事件源）Observable

Observable 被定义成为一个函数，该函数含有一个类型为*Observer的参数。

type Observable func(*Observer) error

任何事件源都是这样的一个函数，当调用该函数即意味着 订阅 了该事件源，入参为一个Observer，具体功能见下面

如果该函数返回nil，即意味着 事件流完成

否则意味着 事件流异常

观察者对象Observer

type Stop chan bool
type Observer struct {
    next     NextHandler //缓存当前的NextHandler，后续可以被替换
    dispose  Stop        //取消订阅的信号，只用来close
    complete Stop        //用于发出完成信号
    err      error       //缓存当前的错误
}

该控制器为一个结构体，其中next记录了当前的NextHandler，

在任何时候，如果关闭了dispose这个channel，就意味着 取消订阅 。

//Dispose 取消订阅
func (c *Observer) Dispose() {
    select {
    case <-c.dispose:
    default:
        close(c.dispose)
    }
}

//Aborted 判断是否已经取消订阅或者已完成
func (c *Observer) Aborted() bool {
    select {
    case <-c.dispose:
        return true
    case <-c.complete:
        return true
    default:
        return false
    }
}

由于Channel的close可以引发所有读取该Channel的阻塞行为唤醒，所以可以在不同层级复用该channel

并且，由于已经close的channel可以反复读取以取得是否close的状态信息，所以不需要再额外记录

Observer对象为Observable和事件处理逻辑共同持有，是二者沟通的桥梁

NextHandler

type Event struct {
    Data    interface{}
    Target  *Observer
}
NextHandler interface {
    OnNext(*Event)
}

NextHandler是一个接口，实现OnNext函数，当Observable数据推送到Observer中时，即调用了该函数

Target属性用于存储当前发送事件的Observer对象，有两大重要使命

1. 更换NextHandler，用于减少数据传递过程
2. 在NextHandler过程中终止事件流

这样做的好处是可以实现不同的观察者，比如函数或者channel

type(
    NextFunc func(*Event)
    NextChan chan *Event
)
func (next NextFunc) OnNext(event *Event) {
    next(event)
}
func (next NextChan) OnNext(event *Event) {
    next <- event
}

实现案例TakeUntil

//TakeUntil 一直获取事件直到unitl传来事件为止
func (ob Observable) TakeUntil(until Observable) Observable {
    return func(sink *Observer) error {
        go until(sink.New3(NextFunc(func(event *Event) {
            //获取到任何数据就让下游完成
            sink.Complete() //由于复用了complete信号，所以会导致所有复用complete的事件流完成
        })))
        return ob(sink)
    }
}

TakeUnitl的用途是，传入一个until事件源，当这个until事件源接受到事件时，就会导致当前的事件源”完成”。相当于某种中断信号。

看似简短的代码，确考虑各种不同的情形

几大实现细节：

1. 订阅until事件源，通过go关键字创建goroutine防止阻塞当前goroutine
2. 使用函数式NextHandler，用户接受来自until事件源的事件，一旦接受任何事件，就调用sink.Complete()来使得当前事件流完成
3. 订阅until事件源的Observer（sink.New3创建)复用了sink.dispose和sink.complete两个信号,当用户在代码中取消了订阅，就会引发该until事件源的取消订阅行为
4. 最后一步是订阅上游事件源，我们不创建新的Observer，而直接把下游的Observer传入，避免了不必要的转发逻辑
5. 任何情况取消订阅，或者上游事件源完成都可以使得事件源函数返回，接着TakeUntil函数也会返回,即意味着完成
6. until事件源的完成或者错误，都将忽略，所以我们没有去获取until函数返回值