rename ingest to pipeline

pipeline.go

@@ -3,7 +3,7 @@ package main
 import "context"
 
 type (
-	Ingester struct {
+	Pipeline struct {
 		writer  Queue
 		reader  Queue
 		process processFunc
@@ -11,31 +11,31 @@ type (
 	processFunc func(context.Context, []byte) ([]byte, error)
 )
 
-func NewIngester(writer, reader Queue, process processFunc) Ingester {
-	return Ingester{
+func NewPipeline(writer, reader Queue, process processFunc) Pipeline {
+	return Pipeline{
 		writer:  writer,
 		reader:  reader,
 		process: process,
 	}
 }
 
-func (i Ingester) Process(ctx context.Context) error {
+func (p Pipeline) Process(ctx context.Context) error {
 	ctx, can := context.WithCancel(ctx)
 	defer can()
 
 	for ctx.Err() == nil {
-		reservation, read, err := i.reader.Syn(ctx)
+		reservation, read, err := p.reader.Syn(ctx)
 		if err != nil {
 			return err
 		}
-		processed, err := i.process(ctx, read)
+		processed, err := p.process(ctx, read)
 		if err != nil {
 			return err
 		}
-		if err := i.writer.Enqueue(ctx, processed); err != nil {
+		if err := p.writer.Enqueue(ctx, processed); err != nil {
 			return err
 		}
-		if err := i.reader.Ack(ctx, reservation); err != nil {
+		if err := p.reader.Ack(ctx, reservation); err != nil {
 			return err
 		}
 	}

pipeline_test.go

@@ -6,7 +6,7 @@ import (
 	"time"
 )
 
-func TestIngester(t *testing.T) {
+func TestPipeline(t *testing.T) {
 	ctx, can := context.WithTimeout(context.Background(), time.Second*10)
 	defer can()
 
@@ -30,7 +30,7 @@ func TestIngester(t *testing.T) {
 		t.Error(err)
 	}
 
-	ing := NewIngester(output, input, process)
+	ing := NewPipeline(output, input, process)
 	go func() {
 		defer can()
 		if err := ing.Process(ctx); err != nil && ctx.Err() == nil {

slack.go

@@ -0,0 +1,14 @@
+package main
+
+import (
+	"context"
+	"io"
+)
+
+type SlackIngestion struct {
+	slackToMessage Pipeline
+}
+
+func NewSlackIngestion(ctx context.Context, driver Driver) (SlackIngestion, error) {
+	return SlackIngestion{}, io.EOF
+}