RxGoGo 语言的 Reactive 扩展

联合创作 · 2023-09-28 12:26

RxGo 是 Go 语言的 Reactive 扩展。

安装

go get -u github.com/jochasinga/rxgo

用法

watcher := observer.Observer{

    // Register a handler function for every next available item.
    NextHandler: func(item interface{}) {
        fmt.Printf("Processing: %v\n", item)
    },

    // Register a handler for any emitted error.
    ErrHandler: func(err error) {
        fmt.Printf("Encountered error: %v\n", err)
    },

    // Register a handler when a stream is completed.
    DoneHandler: func() {
        fmt.Println("Done!")
    },
}

it, _ := iterable.New([]interface{}{1, 2, 3, 4, errors.New("bang"), 5})
source := observable.From(it)
sub := source.Subscribe(watcher)

// wait for the async operation
<-sub

以上将:

  • 将切片中每个数字的格式字符串 print 为4。

  • print 错误“bang”

重要的是要记住,只有一个 OnError 或 OnDone 可以在 stream 中调用。 如果 stream 中有错误,处理停止,OnDone 将永远不会被调用,反之亦然。

概念是将所有“side effects”分组到这些处理程序中,让一个 Observer 或任何 EventHandler 处理它们。

package main
import (
    "fmt"
    "time"

    "github.com/jochasinga/rx"
    "github.com/jochasinga/rx/handlers"
)

func main() {

    score := 9

    onNext := handlers.NextFunc(func(item interface{}) {
        if num, ok := item.(int); ok {
            score += num
        }
    })

    onDone := handlers.DoneFunc(func() {
        score *= 2
    })

    watcher := observer.New(onNext, onDone)

    // Create an `Observable` from a single item and subscribe to the observer.
    sub := observable.Just(1).Subscribe(watcher)
    <-sub

    fmt.Println(score) // 20
}
浏览 6
点赞
评论
收藏
分享

手机扫一扫分享

编辑 分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

编辑 分享
举报