【ReactiveX】基于Golang pmlpml/RxGo程序包的二次开发

冷不防 2022-12-19 13:15 160阅读 0赞

### 基于Golang pmlpml/RxGo程序包的二次开发【阅读时间：约20分钟】 ###

*  一、ReactiveX & RxGo介绍
 *   *  1.ReactiveX
     *  2.RxGo
 *  二、系统环境&项目介绍
 *   *  1.系统环境
     *  2.项目的任务要求
 *  三、具体程序设计及Golang代码实现
 *   *  1.程序设计
     *  2.filteringOperator数据结构、op函数与newFilterObservable函数
     *  3. Debounce函数
     *  4. Distinct函数
     *  5. ElementAt函数
     *  6. Filter函数
     *  7. First函数
     *  8. IgnoreElements函数
     *  9. Last函数
     *  10. Sample函数
     *  11. Skip函数
     *  12. SkipLast函数
     *  13. Take函数
     *  14. TakeLast函数
 *  四、程序测试
 *   *  1.封装并使用程序包
     *  2.功能测试
     *  3.单元测试
 *  五、中文 api 文档
 *  六、完整代码
 *  七、References

# 一、ReactiveX & RxGo介绍 #

## 1.ReactiveX ##

ReactiveX是Reactive Extensions的缩写，一般简写为Rx，最初是LINQ的一个扩展，由微软的架构师Erik Meijer领导的团队开发，在2012年11月开源，Rx是一个编程模型，目标是提供一致的编程接口，帮助开发者更方便的处理异步数据流，Rx库支持.NET、JavaScript和C++，Rx近几年越来越流行了，现在已经支持几乎全部的流行编程语言了，Rx的大部分语言库由ReactiveX这个组织负责维护，比较流行的有RxJava/RxJS/Rx.NET，社区网站是 [reactivex.io][]。[中文文档][Link 1]

## 2.RxGo ##

Go 语言的 [RxGo][] 看上去就像 go 入门不久的人写的，很怪异。 但 RxJava 库写的很好。

[pmlpml/RxGo][pmlpml_RxGo] 模仿 Java 版写了 Go 版本新实现，已基本实现了 Creating Observables 和 Transforming Observables 两类算子。

# 二、系统环境&项目介绍 #

## 1.系统环境 ##

操作系统：CentOS7  
硬件信息：使用virtual box配置虚拟机（内存3G、磁盘30G）  
编程语言：GO 1.15.2

## 2.项目的任务要求 ##

阅读 ReactiveX 文档。请在 [pmlpml/RxGo][pmlpml_RxGo] 基础上，

1.  修改、改进它的实现
2.  或添加一组新的操作，如 [filtering][]

该库的基本组成：

`rxgo.go` 给出了基础类型、抽象定义、框架实现、Debug工具等

`generators.go` 给出了 sourceOperater 的通用实现和具体函数实现

`transforms.go` 给出了 transOperater 的通用实现和具体函数实现

# 三、具体程序设计及Golang代码实现 #

## 1.程序设计 ##

在本次二次开发中，笔者选择的是第二种实现，即添加一组filtering操作。  
在[filtering][]中，我们可以看到该算子的定义及操作如下：  
Filtering Observables: Operators that selectively emit items from a source Observable.

*  [**`Debounce`**][Debounce] — only emit an item from an Observable if a particular timespan has passed without it emitting another item
 *  [**`Distinct`**][Distinct] — suppress duplicate items emitted by an Observable
 *  [**`ElementAt`**][ElementAt] — emit only item *n* emitted by an Observable
 *  [**`Filter`**][Filter] — emit only those items from an Observable that pass a predicate test
 *  [**`First`**][First] — emit only the first item, or the first item that meets a condition, from an Observable
 *  [**`IgnoreElements`**][IgnoreElements] — do not emit any items from an Observable but mirror its termination notification
 *  [**`Last`**][Last] — emit only the last item emitted by an Observable
 *  [**`Sample`**][Sample] — emit the most recent item emitted by an Observable within periodic time intervals
 *  [**`Skip`**][Skip] — suppress the first *n* items emitted by an Observable
 *  [**`SkipLast`**][SkipLast] — suppress the last *n* items emitted by an Observable
 *  [**`Take`**][Take] — emit only the first *n* items emitted by an Observable
 *  [**`TakeLast`**][TakeLast] — emit only the last *n* items emitted by an Observable

## 2.filteringOperator数据结构、op函数与newFilterObservable函数 ##

首先直接借鉴pmlpml/RxGo中的transforms.go文件，完善filteringOperator数据结构、op函数与newFilterObservable函数如下：

// filtering node implementation of streamOperator
    type filteringOperator struct { 
    	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool)
    }
    
    //op op函数
    func (sop filteringOperator) op(ctx context.Context, o *Observable) { 
    	// must hold defintion of flow resourcs here, such as chan etc., that is allocated when connected
    	// this resurces may be changed when operation routine is running.
    	in := o.pred.outflow
    	out := o.outflow
    	//fmt.Println(o.name, "operator in/out chan ", in, out)
    
    	// Scheduler
    	go func() { 
    		end := false
    
    		for x := range in { 
    			if end { 
    				break
    			}
    
    			// can not pass a interface as parameter (pointer) to gorountion for it may change its value outside!
    			temp := reflect.ValueOf(x)
    			// send an error to stream if the flip not accept error
    			err, ok := x.(error)
    			if ok && !o.flip_accept_error { 
    				o.sendToFlow(ctx, err, out)
    				continue
    			}
    
    			if sop.opFunc(ctx, o, temp, out) { 
    				end = true
    			}
    		}
    
    		if o.flip != nil { 
    			buffer := (reflect.ValueOf(o.flip))
    			for i := 0; i < buffer.Len(); i++ { 
    				o.sendToFlow(ctx, buffer.Index(i).Interface(), out)
    			}
    		}
    
    		o.closeFlow(out)
    	}()
    }
    
    //newFilterObservable 新建一个Filter Observabl
    func (parent *Observable) newFilterObservable(name string) (o *Observable) { 
    	//new Observable
    	o = newObservable()
    	o.Name = name
    
    	//chain Observables
    	parent.next = o
    	o.pred = parent
    	o.root = parent.root
    
    	//set options
    	o.buf_len = BufferLen
    	return o
    }

## 3. Debounce函数 ##

Debounce函数用于按时间防抖动，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var count = 0
    var tempTime time.Duration
    
    //Debounce 按时间防抖动
    func (parent *Observable) Debounce(timespan time.Duration) (o *Observable) { 
    	tempTime = timespan
    	o = parent.newFilterObservable("debounce")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    	o.operator = debounceOp
    	return o
    }
    
    var debounceOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		count++
    		go func() { 
    			tempCount := count
    			time.Sleep(tempTime)
    			select { 
    			case <-ctx.Done():
    				return
    			default:
    				if tempCount == count { 
    					o.sendToFlow(ctx, item.Interface(), out)
    				}
    			}
    		}()
    		return false
    	},
    }

## 4. Distinct函数 ##

Distinct函数用于过滤掉重复出现的元素，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var tempMap map[string]bool
    
    //Distinct 过滤掉重复出现的元素
    func (parent *Observable) Distinct() (o *Observable) { 
    	o = parent.newFilterObservable("distinct")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	tempMap = map[string]bool{ }
    	o.operator = distinctOp
    	return o
    }
    
    var distinctOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		itemStr := fmt.Sprintf("%v", item)
    		_, ok := tempMap[itemStr]
    		if !ok { 
    			tempMap[itemStr] = true
    			o.sendToFlow(ctx, item.Interface(), out)
    		}
    		return false
    	},
    }

## 5. ElementAt函数 ##

ElementAt函数用于取第几个元素，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var tempNum int
    
    //ElementAt 取第几个元素
    func (parent *Observable) ElementAt(num int) (o *Observable) { 
    	tempNum = num
    	o = parent.newFilterObservable("elementAt")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    	o.operator = elementAtOp
    
    	return o
    }
    
    var elementAtOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		if count == tempNum { 
    			o.sendToFlow(ctx, item.Interface(), out)
    			return true
    		}
    		count++
    		return false
    	},
    }

## 6. Filter函数 ##

Filter函数用于过滤出特定的数据，此处在transforms.go文件中已定义该操作，直接将其复制到filtering.go文件中即可，另外此处笔者为了避免函数冲突，将transforms.go中的filter函数改为了filter2函数。Observable.operator等于具体的算子操作即可，其具体代码实现如下：

// Filter `func(x anytype) bool` filters items in the original Observable and returns
    // a new Observable with the filtered items.
    func (parent *Observable) Filter(f interface{ }) (o *Observable) { 
    	// check validation of f
    	fv := reflect.ValueOf(f)
    	inType := []reflect.Type{ typeAny}
    	outType := []reflect.Type{ typeBool}
    	b, ctx_sup := checkFuncUpcast(fv, inType, outType, true)
    	if !b { 
    		panic(ErrFuncFlip)
    	}
    
    	o = parent.newTransformObservable("filter")
    	o.flip_accept_error = checkFuncAcceptError(fv)
    
    	o.flip_sup_ctx = ctx_sup
    	o.flip = fv.Interface()
    	o.operator = filterOperater
    	return o
    }
    
    var filterOperater = transOperater{ func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{ }) (end bool) { 
    
    	fv := reflect.ValueOf(o.flip)
    	var params = []reflect.Value{ x}
    	rs, skip, stop, e := userFuncCall(fv, params)
    
    	var item interface{ } = rs[0].Interface()
    	if stop { 
    		end = true
    		return
    	}
    	if skip { 
    		return
    	}
    	if e != nil { 
    		item = e
    	}
    	// send data
    	if !end { 
    		if b, ok := item.(bool); ok && b { 
    			end = o.sendToFlow(ctx, x.Interface(), out)
    		}
    	}
    
    	return
    }}

## 7. First函数 ##

First函数用于完成时返回第一个元素，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

//First 完成时返回第一个元素
    func (parent *Observable) First() (o *Observable) { 
    	o = parent.newFilterObservable("first")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	o.operator = firstOp
    	return o
    }
    
    var firstOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		o.sendToFlow(ctx, item.Interface(), out)
    		return true
    	},
    }

## 8. IgnoreElements函数 ##

函数用于，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

## 9. Last函数 ##

Last函数用于完成时返回最后一个元素，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

//Last 完成时返回最后一个元素
    func (parent *Observable) Last() (o *Observable) { 
    	o = parent.newFilterObservable("last")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	o.operator = lastOp
    	return o
    }
    
    var lastOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		o.flip = append([]interface{ }{ }, item.Interface())
    		return false
    	},
    }

## 10. Sample函数 ##

Sample函数用于定期发射Observable最近发射的数据项，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var tempSample chan interface{ }
    
    //Sample 定期发射Observable最近发射的数据项
    func (parent *Observable) Sample(sample chan interface{ }) (o *Observable) { 
    	tempSample = sample
    	o = parent.newFilterObservable("sample")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	o.operator = sampleOP
    	return o
    }
    
    var sampleOP = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		var latest interface{ } = nil
    		latest = item.Interface()
    		go func() { 
    			tempEnd := true
    			for tempEnd { 
    				select { 
    				case <-ctx.Done():
    					tempEnd = true
    				case <-tempSample:
    					if latest != nil { 
    						if o.sendToFlow(ctx, latest, out) { 
    							tempEnd = false
    						}
    						latest = nil
    					}
    				}
    			}
    		}()
    		return false
    	},
    }

## 11. Skip函数 ##

Skip函数用于跳过前n个数据，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

//Skip 跳过前n个数据
    func (parent *Observable) Skip(num int) (o *Observable) { 
    	tempNum = num
    	o = parent.newFilterObservable("skip")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    	o.operator = skipOp
    
    	return o
    }
    
    var skipOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		count++
    		if count > tempNum { 
    			o.sendToFlow(ctx, item.Interface(), out)
    		}
    		return false
    	},
    }

## 12. SkipLast函数 ##

SkipLast函数用于跳过最后n个数据，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var tempLasts []interface{ }
    
    //SkipLast 跳过最后n个数据
    func (parent *Observable) SkipLast(num int) (o *Observable) { 
    	tempNum = num
    	o = parent.newFilterObservable("skipLast")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    	o.operator = skipLastOp
    
    	return o
    }
    
    var skipLastOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		//var lasts []interface{}
    		if count == tempNum { 
    			o.sendToFlow(ctx, tempLasts[0], out)
    			tempLasts = tempLasts[1:]
    		} else { 
    			count++
    		}
    		tempLasts = append(tempLasts, item.Interface())
    		return false
    	},
    }

## 13. Take函数 ##

Take函数用于取前n个数据，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

//Take 取前n个数据
    func (parent *Observable) Take(num int) (o *Observable) { 
    	tempNum = num
    	o = parent.newFilterObservable("take")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    	o.operator = takeOp
    
    	return o
    }
    
    var takeOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		count++
    		if count > tempNum { 
    			return true
    		}
    		o.sendToFlow(ctx, item.Interface(), out)
    		return false
    	},
    }

## 14. TakeLast函数 ##

TakeLast函数用于取最后n个数据，借鉴transforms.go文件的实现方式，Observable.operator等于具体的算子操作即可，其具体代码实现如下：

var tempLasts2 []interface{ }
    
    //TakeLast 取最后n个数据
    func (parent *Observable) TakeLast(num int) (o *Observable) { 
    	tempNum = num
    	o = parent.newFilterObservable("takeLast")
    	o.flip_accept_error = true
    	o.flip_sup_ctx = true
    	count = 0
    
    	o.operator = takeLastOp
    
    	return o
    }
    
    var takeLastOp = filteringOperator{ 
    	opFunc: func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{ }) (end bool) { 
    		count++
    		if count <= tempNum { 
    			tempLasts2 = append(tempLasts2, item.Interface())
    		} else { 
    			tempLasts2 = tempLasts2[1:]
    			tempLasts2 = append(tempLasts2, item.Interface())
    		}
    		o.flip = tempLasts2
    		return false
    	},
    }

# 四、程序测试 #

## 1.封装并使用程序包 ##

在项目rxgo的目录下，执行如下指令:

go build

在其他路径下建立main.go，并调用rxgo库即可。

## 2.功能测试 ##

功能测试主要从用户角度测试程序包的功能，步骤如下:  
创建main.go文件，内容如下（数值可自定义）:

package main
    
    import (
    	"fmt"
    	"time"
    
    	"github.com/user/rxgo"
    )
    
    func main() { 
    	fmt.Print("Debounce: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Debounce(2).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("Distinct: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Distinct().Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("ElementAt 3: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).ElementAt(3).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("First: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).First().Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	//filter
    	fmt.Print("Filter value < 4: ")
    	res := []int{ }
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Filter(func(x int) bool { 
    		return x < 4
    	}).Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	for i := range res { 
    		fmt.Print(res[i])
    		fmt.Print(" ")
    	}
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("IgnoreElements: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).IgnoreElements().Subscribe(func(x int) { 
    		fmt.Print(x)
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("Last: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Last().Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("Sample: ")
    	observableP := make(chan interface{ })
    	go func() { 
    		rxgo.Just(1, 2).Map(func(x int) int { 
    			switch x { 
    			case 1:
    				time.Sleep(10 * time.Millisecond)
    			case 2:
    				time.Sleep(5 * time.Millisecond)
    			default:
    				time.Sleep(10 * time.Millisecond)
    			}
    			return x
    		}).Subscribe(func(x int) { 
    			observableP <- x
    		})
    	}()
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Map(func(x int) int { 
    		time.Sleep(3 * time.Millisecond)
    		return x
    	}).Sample(observableP).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("Skip 2: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Skip(2).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("SkipLast 2: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).SkipLast(2).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("Take 4: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Take(4).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    	fmt.Print("TakeLast 3: ")
    	rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).TakeLast(3).Subscribe(func(x int) { 
    		fmt.Print(x)
    		fmt.Print(" ")
    	})
    	fmt.Println()
    	fmt.Println()
    
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ3OTA5Mg_size_16_color_FFFFFF_t_70_pic_center]  
由此可知程序包的功能测试结果正确，符合的程序包中关于filtering操作的定义。

## 3.单元测试 ##

单元测试主要从程序员角度，对程序包的具体函数进行测试。  
建立filtering\_test.go文件，对程序包的每个函数进行单元测试，此处同样借鉴pmlpml/RxGo中的transform\_test文件，内容如下:

package rxgo_test
    
    import (
    	"testing"
    
    	"github.com/stretchr/testify/assert"
    	"github.com/user/rxgo"
    )
    
    func TestDebounce(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Map(func(x int) int { 
    		return x
    	}).Debounce(1)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    
    	assert.Equal(t, []int{ 1}, res, "Debounce Test Errorr")
    }
    
    func TestDistinct(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 2, 6).Map(func(x int) int { 
    		return x
    	}).Distinct()
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 1, 8, 3, 4, 2, 0, 6}, res, "Distinct Test Errorr")
    }
    
    func TestElementAt(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).ElementAt(2)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 3}, res, "SkipLast Test Errorr")
    }
    
    func TestFirst(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).First()
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    
    	assert.Equal(t, []int{ 1}, res, "First Test Errorr")
    }
    
    func TestIgnoreElements(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).IgnoreElements()
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ }, res, "IgnoreElementsTest Errorr")
    }
    
    func TestLast(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).Last()
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    
    	assert.Equal(t, []int{ 6}, res, "Last Test Errorr")
    }
    
    func TestSkip(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).Skip(3)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 4, 2, 0, 6}, res, "Skip Test Errorr")
    }
    
    func TestSkipLast(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).SkipLast(3)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 1, 8, 3, 4}, res, "SkipLast Test Errorr")
    }
    
    func TestTake(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).Take(4)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 1, 8, 3, 4}, res, "Take Test Errorr")
    }
    
    func TestTakeLast(t *testing.T) { 
    	res := []int{ }
    	ob := rxgo.Just(1, 8, 3, 4, 2, 0, 6).Map(func(x int) int { 
    		return x
    	}).TakeLast(4)
    	ob.Subscribe(func(x int) { 
    		res = append(res, x)
    	})
    	assert.Equal(t, []int{ 4, 2, 0, 6}, res, "TakeLast Test Errorr")
    }

结果如下（出于篇幅考虑，部分省略，用…代替）：

[henryhzy@localhost rxgo]$ go test -v
    === RUN   TestDebounce
    --- PASS: TestDebounce (0.00s)
    === RUN   TestDistinct
    --- PASS: TestDistinct (0.00s)
    === RUN   TestElementAt
    --- PASS: TestElementAt (0.00s)
    === RUN   TestFirst
    --- PASS: TestFirst (0.00s)
    === RUN   TestIgnoreElements
    --- PASS: TestIgnoreElements (0.00s)
    === RUN   TestLast
    --- PASS: TestLast (0.00s)
    === RUN   TestSkip
    --- PASS: TestSkip (0.00s)
    === RUN   TestSkipLast
    --- PASS: TestSkipLast (0.00s)
    === RUN   TestTake
    --- PASS: TestTake (0.00s)
    === RUN   TestTakeLast
    --- PASS: TestTakeLast (0.00s)
    ...
    ...
    ok  	github.com/user/rxgo	0.005s
    [henryhzy@localhost rxgo]$

# 五、中文 api 文档 #

首先安装godoc如下:

git clone https://github.com/golang/tools $GOPATH/src/golang.org/x/tools
    go build golang.org/x/tools

在项目rxgo所在目录下，执行如下指令:

go install
    go doc
    godoc -url="pkg/github.com/user/rxgo" > API.html

由此可知程序包的单元测试结果正确，符合的程序包中关于filtering操作的定义。

# 六、完整代码 #

具体代码可见gitee仓库：[gitee][]

# 七、References #

1.  [课程博客][Link 2]
2.  [ReactiveX官网][reactivex.io]

[reactivex.io]: http://reactivex.io/documentation/observable.html
[Link 1]: https://mcxiaoke.gitbooks.io/rxdocs/content/Intro.html
[RxGo]: https://github.com/ReactiveX/RxGo
[pmlpml_RxGo]: https://github.com/pmlpml/rxgo
[filtering]: http://reactivex.io/documentation/operators.html#filtering
[Debounce]: http://reactivex.io/documentation/operators/debounce.html
[Distinct]: http://reactivex.io/documentation/operators/distinct.html
[ElementAt]: http://reactivex.io/documentation/operators/elementat.html
[Filter]: http://reactivex.io/documentation/operators/filter.html
[First]: http://reactivex.io/documentation/operators/first.html
[IgnoreElements]: http://reactivex.io/documentation/operators/ignoreelements.html
[Last]: http://reactivex.io/documentation/operators/last.html
[Sample]: http://reactivex.io/documentation/operators/sample.html
[Skip]: http://reactivex.io/documentation/operators/skip.html
[SkipLast]: http://reactivex.io/documentation/operators/skiplast.html
[Take]: http://reactivex.io/documentation/operators/take.html
[TakeLast]: http://reactivex.io/documentation/operators/takelast.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjQ3OTA5Mg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/fe81f5c0f51d4dfe8c4b22f6782d99d7.png
[gitee]: https://gitee.com/HenryHZY/sysu-service-computing/tree/master/hw6-rxgo
[Link 2]: https://pmlpml.gitee.io/service-computing/post/ex-pkg-rxgo/