fix: issue ReactiveX#341

item.go

@@ -84,6 +84,39 @@ func send(ctx context.Context, ch chan<- Item, items ...interface{}) {
  }
 }
 
+func sendSingleItem(ctx context.Context, ch chan<- Item, strategy CloseChannelStrategy, items ...interface{}) {
+ if strategy == CloseChannel {
+ defer close(ch)
+ }
+ for _, currentItem := range items {
+ switch item := currentItem.(type) {
+ default:
+ rt := reflect.TypeOf(item)
+ switch rt.Kind() {
+ default:
+ Of(item).SendContext(ctx, ch)
+ case reflect.Chan:
+ in := reflect.ValueOf(currentItem)
+ for {
+ v, ok := in.Recv()
+ if !ok {
+ return
+ }
+ currentItem := v.Interface()
+ switch item := currentItem.(type) {
+ default:
+ Of(item).SendContext(ctx, ch)
+ case error:
+ Error(item).SendContext(ctx, ch)
+ }
+ }
+ }
+ case error:
+ Error(item).SendContext(ctx, ch)
+ }
+ }
+}
+
 // Error checks if an item is an error.
 func (i Item) Error() bool {
  return i.E != nil

iterable_just.go

@@ -18,6 +18,6 @@ func (i *justIterable) Observe(opts ...Option) <-chan Item {
  option := parseOptions(append(i.opts, opts...)...)
  next := option.buildChannel()
 
- go SendItems(option.buildContext(emptyContext), next, CloseChannel, i.items)
+ go sendSingleItem(option.buildContext(emptyContext), next, CloseChannel, i.items...)
  return next
 }

single.go

@@ -1,6 +1,8 @@
 package rxgo
 
-import "context"
+import (
+ "context"
+)
 
 // Single is a observable with a single element.
 type Single interface {