mapOperator missing res when return error in "map" function

[feng7208485]

here is my code

func main() {
    supplier := make([]rxgo.Producer, 0)
    supplier = append(supplier, GenCheckFieldTypeFuture(context.Background(), AAA{Foo: "111"}))
    observable := rxgo.Create(supplier).Map(func(_ context.Context, i interface{}) (interface{}, error) {
        if v, ok := i.(CheckFieldTypeParam); ok {
            println(fmt.Sprintf("v type is : %v", reflect.TypeOf(v)))
            return v, errors.New("asasfqwfq")
        }
        return i, nil
    }, rxgo.WithPool(1))

    for item := range observable.Observe() {
        println(fmt.Sprintf("item.V type is:%v", reflect.TypeOf(item.V)))
        println(fmt.Sprintf("item.E type is:%v", reflect.TypeOf(item.E)))
    }
}

the out put is

v type is : main.CheckFieldTypeParam
item.V type is:<nil>
item.E type is:*errors.errorString

Expected behavior I expect item.V is not nil, the framwork should not drop result even if the map fun return an error!!!

[feng7208485]

this bug should fix by following code:

func (op *mapOperator) next(ctx context.Context, item Item, dst chan<- Item, operatorOptions operatorOptions) {
    res, err := op.apply(ctx, item.V)
    if err != nil {
        Of(res).Error(err).SendContext(ctx,** dst)
        operatorOptions.stop()
        return
    }
    Of(res).SendContext(ctx, dst)
}

just add "Of(res)" before line 4