add opt (#6)

Signed-off-by: clavinjune <24659468+clavinjune@users.noreply.github.com>

Signed-off-by: clavinjune <24659468+clavinjune@users.noreply.github.com>

example/simple.go

@@ -65,10 +65,16 @@ func main() {
 		"name": "piper simple example",
 	}
 
+	popt := &piper.Opt{
+		Context:     ctx,
+		Data:        m,
+		TotalWorker: totalWorker,
+		FilterError: true,
+	}
 	// create pipeline functions
-	addOneFn := piper.New(ctx, totalWorker, m, add65)
-	toRunFn := piper.New(ctx, totalWorker, m, toRune)
-	toStringFn := piper.New(ctx, totalWorker, m, toString)
+	addOneFn := piper.New(add65, popt)
+	toRunFn := piper.New(toRune, popt)
+	toStringFn := piper.New(toString, popt)
 
 	start := time.Now()
 

opt.go

@@ -0,0 +1,35 @@
+// Copyright 2022 clavinjune/piper
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package piper
+
+import (
+	"context"
+)
+
+// Opt holds options for P
+type Opt struct {
+	_ struct{}
+	// Context will be passed over each pipeline
+	Context context.Context
+
+	// Data will be converted to *sync.Map
+	Data map[string]any
+
+	// TotalWorker default 1
+	TotalWorker int
+
+	// FilterError will skip channel with error if enabled
+	FilterError bool
+}

pipeline.go

@@ -19,9 +19,18 @@ import (
 	"sync"
 )
 
+const (
+	defaultTotalWorker = 1
+)
+
+var (
+	defaultContext = context.Background()
+)
+
 // P is the pipeline
 type P[IN, OUT any] struct {
 	_           struct{}
+	filterError bool
 	ctx         context.Context
 	totalWorker int
 	data        *sync.Map
@@ -66,6 +75,9 @@ ChannelReader:
 
 func (p *P[IN, OUT]) workDefaultAction(n *W[IN], out chan<- *W[OUT]) {
 	if n.Err != nil {
+		if p.filterError {
+			return
+		}
 		out <- &W[OUT]{
 			Err: n.Err,
 		}
@@ -77,23 +89,51 @@ func (p *P[IN, OUT]) workDefaultAction(n *W[IN], out chan<- *W[OUT]) {
 		In:  n.Data,
 	})
 
+	if err != nil && p.filterError {
+		return
+	}
+
 	out <- &W[OUT]{
 		Data: o,
 		Err:  err,
 	}
 }
 
 // New creates new pipeline
-func New[IN, OUT any](ctx context.Context, totalWorker int, data map[string]any, fn F[IN, OUT]) *P[IN, OUT] {
-	d := new(sync.Map)
-	for k, v := range data {
-		d.Store(k, v)
+func New[IN, OUT any](fn F[IN, OUT], opts ...*Opt) *P[IN, OUT] {
+	p := &P[IN, OUT]{
+		fn: fn,
 	}
 
-	return &P[IN, OUT]{
-		ctx:         ctx,
-		totalWorker: totalWorker,
-		data:        d,
-		fn:          fn,
+	if len(opts) == 0 {
+		return p
 	}
+
+	p.fill(opts[0])
+	if p.totalWorker == 0 {
+		p.totalWorker = defaultTotalWorker
+	}
+
+	if p.ctx == nil {
+		p.ctx = defaultContext
+	}
+
+	return p
+}
+
+func (p *P[IN, OUT]) fill(o *Opt) {
+	if o == nil {
+		return
+	}
+
+	p.ctx = o.Context
+	p.totalWorker = o.TotalWorker
+	p.filterError = o.FilterError
+
+	m := new(sync.Map)
+	for k, v := range o.Data {
+		m.Store(k, v)
+	}
+
+	p.data = m
 }