jjeffcaii / reactor-go

A golang implementation for reactive-streams.

reactor-go 🚀🚀🚀

A golang implementation for reactive-streams.

Install

go get -u github.com/jjeffcaii/reactor-go

Example

Mono

package mono_test

import (
	"context"
	"fmt"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/mono"
)

func Example() {
	gen := func(ctx context.Context, sink mono.Sink) {
		sink.Success("World")
	}
	mono.
		Create(gen).
		Map(func(input reactor.Any) (output reactor.Any, err error) {
			output = "Hello " + input.(string) + "!"
			return
		}).
		DoOnNext(func(v reactor.Any) error {
			fmt.Println(v)
			return nil
		}).
		Subscribe(context.Background())
}

// Should print
// Hello World!

Flux

package flux_test

import (
	"context"
	"fmt"

	"github.com/jjeffcaii/reactor-go"
	"github.com/jjeffcaii/reactor-go/flux"
	"github.com/jjeffcaii/reactor-go/scheduler"
)

func Example() {
	gen := func(ctx context.Context, sink flux.Sink) {
		for i := 0; i < 10; i++ {
			v := i
			sink.Next(v)
		}
		sink.Complete()
	}
	done := make(chan struct{})

	var su reactor.Subscription
	flux.Create(gen).
		Filter(func(i interface{}) bool {
			return i.(int)%2 == 0
		}).
		Map(func(input reactor.Any) (output reactor.Any, err error) {
			output = fmt.Sprintf("#HELLO_%04d", input.(int))
			return
		}).
		SubscribeOn(scheduler.Elastic()).
		Subscribe(context.Background(),
			reactor.OnSubscribe(func(s reactor.Subscription) {
				su = s
				s.Request(1)
			}),
			reactor.OnNext(func(v reactor.Any) error {
				fmt.Println("next:", v)
				su.Request(1)
				return nil
			}),
			reactor.OnComplete(func() {
				close(done)
			}),
		)
	<-done
}
// Should print:
// next: #HELLO_0000
// next: #HELLO_0002
// next: #HELLO_0004
// next: #HELLO_0006
// next: #HELLO_0008

reactor-go's Issues

Incorrect calling OnError when using SwitchIfEmpty + Just.

Duplicated 'OnComplete' when using 'SwitchIfEmpty' operator.

Go version (1.18 or 1.18.2) invalid use of type alias FnSwitchOnFirst in recursive type (see issue #50729)

Using go version 1.18 or 1.18.1, seeing the compilation error on recursive type error , not seeing this issue in go lang versions 1.17.9 or older.

invalid use of type alias FnSwitchOnFirst in recursive type in file file types.go under flux pkg.

Please refer to issue which is in open state : golang/go#50729

Is this still mntained?

同学，您这个项目引入了16个开源组件，存在3个漏洞，辛苦升级一下

检测到 jjeffcaii/reactor-go 一共引入了16个开源组件，存在3个漏洞

漏洞标题：Go SSH拒绝服务漏洞
缺陷组件：golang.org/x/[email protected]
漏洞编号：CVE-2020-9283
漏洞描述：Go SSH是一个使用go语言开发的极度简洁的ssh工具，用于远程管理linux、unix等机器。
Go SSH存在拒绝服务漏洞，该漏洞源于网络系统或产品未对输入的数据进行正确的验证，攻击者可利用该漏洞导致拒绝服务条件，拒绝向合法用户提供服务。
国家漏洞库信息：https://www.cnvd.org.cn/flaw/show/CNVD-2020-14300
影响范围：(∞, 0.0.0-20200220183623-bac4c82f6975)
最小修复版本：0.0.0-20200220183623-bac4c82f6975
缺陷组件引入路径：github.com/jjeffcaii/reactor-go@->golang.org/x/[email protected]

另外还有3个漏洞，详细报告：https://mofeisec.com/jr?p=a8c148

Reflection instead of type conversion?

Most operation have func(interface{})interface{} or func(interface{}), as noted in the README. That go me thinking of the possible alternatives and I present the following.

Instead of func(interface{}) we use interface{}, then, using reflection, we make sure that it is a function with inputs and outputs of the correct type. Input of type interface{} is also supported to be backwards compatible with existing code.

Performance wise (TestExample run, average of a few runs):

type conversion: 4176 ns/op
reflect method I: 5472 ns/op
reflect method II: 6458 ns/op

There is room for improvement. Maybe additional optimizations can be done to increase performance.

Here's a demo for the OnNext

// mono/op_peek.go
func (p *peekSubscriber) OnNext(v interface{}) {
	if atomic.LoadInt32(&(p.stat)) != 0 {
		return
	}
	if call := p.parent.onNextCall; call != nil {
		defer func() {
			if err := internal.TryRecoverError(recover()); err != nil {
				p.OnError(err)
			}
		}()

		fn := reflect.ValueOf(call)

		if fn.Kind() != reflect.Func {
			// 'call' is not a function
			return
		}

		// Method I.
		// if fn.Type().NumIn() != 1 {
		// 	// wrong number of inputs
		// }

		// if fn.Type().NumOut() != 0 {
		// 	// wrong number of outputs
		// 	return
		// }

		// // interface is also an acceptable input
		// if fn.Type().In(0).Kind() != reflect.Interface && fn.Type().In(0) != reflect.TypeOf(v) {
		// 	// wrong input kind
		// 	return
		// }

		// fn.Call([]reflect.Value{reflect.ValueOf(v)})

		// Method II.
		var typeFn = reflect.FuncOf([]reflect.Type{reflect.TypeOf(v)}, nil, false)
		var interfaceFn = reflect.FuncOf([]reflect.Type{reflect.TypeOf(new(interface{})).Elem()}, nil, false)

		if !fn.Type().ConvertibleTo(typeFn) && !fn.Type().ConvertibleTo(interfaceFn) {
			// 'call' function is of wrong signature
			return
		}

		fn.Call([]reflect.Value{reflect.ValueOf(v)})
	}
	p.actual.OnNext(v)
}

// test 
func TestExample(t *testing.T) {
	gen := func(ctx context.Context, sink mono.Sink) {
		sink.Success("World")
	}
	mono.
		Create(gen).
		Map(func(i interface{}) interface{} {
			return "Hello " + i.(string) + "!"
		}).
		// note that the function is now a concrete type
		DoOnNext(func(v string) {
			fmt.Println(v)
		}).
		Subscribe(context.Background())
}```