add Take and ToChan for flux.

README.md

@@ -8,7 +8,7 @@
 
 > A golang implementation for [reactive-streams](https://www.reactive-streams.org/).
 <br>🚧🚧🚧 ***IT IS UNDER ACTIVE DEVELOPMENT!!!***
-<br>***[WARNING] DO NOT USE IN ANY PRODUCTION ENVIRONMENT!!!***
+<br>⚠️⚠️⚠️ ***DO NOT USE IN ANY PRODUCTION ENVIRONMENT!!!***
 
 ## Install
 

flux/flux.go

@@ -1,48 +1,50 @@
 package flux
 
 import (
-	"context"
+  "context"
 
-	"github.com/jjeffcaii/reactor-go"
-	"github.com/jjeffcaii/reactor-go/scheduler"
+  "github.com/jjeffcaii/reactor-go"
+  "github.com/jjeffcaii/reactor-go/scheduler"
 )
 
 type OverflowStrategy int8
 
 const (
-	OverflowBuffer OverflowStrategy = iota
-	OverflowIgnore
-	OverflowError
-	OverflowDrop
-	OverflowLatest
+  OverflowBuffer OverflowStrategy = iota
+  OverflowIgnore
+  OverflowError
+  OverflowDrop
+  OverflowLatest
 )
 
 type FnSwitchOnFirst = func(s Signal, f Flux) Flux
 
 type Flux interface {
-	rs.Publisher
-	Filter(rs.Predicate) Flux
-	Map(rs.Transformer) Flux
-	DoOnDiscard(rs.FnOnDiscard) Flux
-	DoOnNext(rs.FnOnNext) Flux
-	DoOnComplete(rs.FnOnComplete) Flux
-	DoOnError(rs.FnOnError) Flux
-	DoOnCancel(rs.FnOnCancel) Flux
-	DoOnRequest(rs.FnOnRequest) Flux
-	DoOnSubscribe(rs.FnOnSubscribe) Flux
-	DoFinally(rs.FnOnFinally) Flux
-	SwitchOnFirst(FnSwitchOnFirst) Flux
-	SubscribeOn(scheduler.Scheduler) Flux
-	BlockLast(context.Context) (interface{}, error)
+  rs.Publisher
+  Filter(rs.Predicate) Flux
+  Map(rs.Transformer) Flux
+  Take(n int) Flux
+  DoOnDiscard(rs.FnOnDiscard) Flux
+  DoOnNext(rs.FnOnNext) Flux
+  DoOnComplete(rs.FnOnComplete) Flux
+  DoOnError(rs.FnOnError) Flux
+  DoOnCancel(rs.FnOnCancel) Flux
+  DoOnRequest(rs.FnOnRequest) Flux
+  DoOnSubscribe(rs.FnOnSubscribe) Flux
+  DoFinally(rs.FnOnFinally) Flux
+  SwitchOnFirst(FnSwitchOnFirst) Flux
+  SubscribeOn(scheduler.Scheduler) Flux
+  BlockLast(context.Context) (interface{}, error)
+  ToChan(ctx context.Context, cap int) (c <-chan interface{}, e <-chan error)
 }
 
 type Sink interface {
-	Complete()
-	Error(error)
-	Next(interface{})
+  Complete()
+  Error(error)
+  Next(interface{})
 }
 
 type Processor interface {
-	Flux
-	Sink
+  Flux
+  Sink
 }

flux/flux_create_sink.go

@@ -1,16 +1,14 @@
 package flux
 
 import (
-	"context"
-	"sync"
-	"sync/atomic"
+  "sync"
+  "sync/atomic"
 
-	rs "github.com/jjeffcaii/reactor-go"
-	"github.com/jjeffcaii/reactor-go/hooks"
+  rs "github.com/jjeffcaii/reactor-go"
+  "github.com/jjeffcaii/reactor-go/hooks"
 )
 
 type bufferedSink struct {
-	ctx      context.Context
 	s        rs.Subscriber
 	q        queue
 	n        int32
@@ -84,7 +82,6 @@ func (p *bufferedSink) drain() {
 func (p *bufferedSink) dispose() {
 	p.done = true
 	_ = p.q.Close()
-
 }
 
 func newBufferedSink(s rs.Subscriber, cap int) *bufferedSink {