From bef06e1a0b77a25c98a729b456e42d3ce714dee4 Mon Sep 17 00:00:00 2001
From: Konstantin Voykov <koykov0@gmail.com>
Date: Fri, 10 Jul 2026 23:13:04 +0300
Subject: [PATCH 1/3] pipeline: feature draft

---
 pipeline.go | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)
 create mode 100644 pipeline.go

diff --git a/pipeline.go b/pipeline.go
new file mode 100644
index 0000000..95c2cbf
--- /dev/null
+++ b/pipeline.go
@@ -0,0 +1,21 @@
+package queue
+
+import "io"
+
+type Pipeline interface {
+	io.Closer
+	Enqueuer
+	AddQueue(q *Queue) Pipeline
+	AddAsyncQueue(q *Queue) Pipeline
+	AddPipeline(p Pipeline) Pipeline
+	AddAsyncPipeline(p Pipeline) Pipeline
+}
+
+type pipeline struct {
+	// ...
+}
+
+func (p *pipeline) Enqueue(x any) error {
+	// todo implement me
+	return nil
+}

From 8b757ac0c95942449897a3d9dee2b51a2a2dbcb8 Mon Sep 17 00:00:00 2001
From: Konstantin Voykov <koykov0@gmail.com>
Date: Sun, 12 Jul 2026 23:34:59 +0300
Subject: [PATCH 2/3] pipeline: rework approach

---
 pipeline.go | 41 +++++++++++++++++++++++++++++++++++++----
 1 file changed, 37 insertions(+), 4 deletions(-)

diff --git a/pipeline.go b/pipeline.go
index 95c2cbf..cdd6653 100644
--- a/pipeline.go
+++ b/pipeline.go
@@ -1,21 +1,54 @@
 package queue
 
-import "io"
+import (
+	"errors"
+	"io"
+)
 
 type Pipeline interface {
 	io.Closer
 	Enqueuer
 	AddQueue(q *Queue) Pipeline
 	AddAsyncQueue(q *Queue) Pipeline
-	AddPipeline(p Pipeline) Pipeline
-	AddAsyncPipeline(p Pipeline) Pipeline
 }
 
 type pipeline struct {
-	// ...
+	buf []pipeq
+}
+
+type pipeq struct {
+	q     *Queue
+	async bool
+}
+
+func NewPipeline() Pipeline {
+	return &pipeline{}
 }
 
 func (p *pipeline) Enqueue(x any) error {
 	// todo implement me
 	return nil
 }
+
+func (p *pipeline) AddQueue(q *Queue) Pipeline {
+	p.buf = append(p.buf, pipeq{q: q})
+	return p
+}
+
+func (p *pipeline) AddAsyncQueue(q *Queue) Pipeline {
+	p.buf = append(p.buf, pipeq{q: q, async: true})
+	return p
+}
+
+func (p *pipeline) Close() error {
+	var errs []error
+	for i := 0; i < len(p.buf); i++ {
+		if err := p.buf[i].q.Close(); err != nil {
+			errs = append(errs, err)
+		}
+	}
+	if len(errs) > 0 {
+		return errors.Join(errs...)
+	}
+	return nil
+}

From bea6a54ddf254cd794173d83f0e718525b8ad5a6 Mon Sep 17 00:00:00 2001
From: Konstantin Voykov <koykov0@gmail.com>
Date: Sun, 12 Jul 2026 23:48:06 +0300
Subject: [PATCH 3/3] pipeline: rework approach v2

---
 pipeline.go | 23 +++++++++++++++++++----
 queue.go    |  4 ++++
 2 files changed, 23 insertions(+), 4 deletions(-)

diff --git a/pipeline.go b/pipeline.go
index cdd6653..b72c0dd 100644
--- a/pipeline.go
+++ b/pipeline.go
@@ -31,13 +31,11 @@ func (p *pipeline) Enqueue(x any) error {
 }
 
 func (p *pipeline) AddQueue(q *Queue) Pipeline {
-	p.buf = append(p.buf, pipeq{q: q})
-	return p
+	return p.add(q, false)
 }
 
 func (p *pipeline) AddAsyncQueue(q *Queue) Pipeline {
-	p.buf = append(p.buf, pipeq{q: q, async: true})
-	return p
+	return p.add(q, true)
 }
 
 func (p *pipeline) Close() error {
@@ -52,3 +50,20 @@ func (p *pipeline) Close() error {
 	}
 	return nil
 }
+
+func (p *pipeline) add(q *Queue, async bool) Pipeline {
+	if q == nil {
+		return p
+	}
+	q.donefn = p.donefn
+	n := len(p.buf)
+	if n > 0 {
+		p.buf[n-1].q.nextq = q
+	}
+	p.buf = append(p.buf, pipeq{q: q, async: async})
+	return p
+}
+
+func (p *pipeline) donefn(x any) {
+	// todo design and implement
+}
diff --git a/queue.go b/queue.go
index 683cd47..77d9eee 100644
--- a/queue.go
+++ b/queue.go
@@ -47,6 +47,10 @@ type Queue struct {
 	// Workers pool.
 	workers []*worker
 
+	nextq *Queue
+	// Item processing done hook.
+	donefn func(any)
+
 	once sync.Once
 
 	// Counter of active workers.