Create core and coretest packages

Signed-off-by: Andrew Hare <[email protected]>

Author: Andrew Hare

.testignore

@@ -1,2 +1,3 @@
 github.com/rename-this/vhs/cmd/vhs
 github.com/rename-this/vhs/internal/smoke
+github.com/rename-this/vhs/coretest

capture/listener.go

@@ -8,7 +8,7 @@ import (
 
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/pcap"
-	"github.com/rename-this/vhs/session"
+	"github.com/rename-this/vhs/core"
 
 	// See https://pkg.go.dev/github.com/google/gopacket?tab=doc#hdr-A_Final_Note
 	_ "github.com/google/gopacket/layers"
@@ -26,7 +26,7 @@ func NewListener(cap *Capture) Listener {
 // given address and port.
 type Listener interface {
 	Packets() <-chan gopacket.Packet
-	Listen(session.Context)
+	Listen(core.Context)
 	Close()
 }
 
@@ -46,9 +46,9 @@ func (l *listener) Packets() <-chan gopacket.Packet {
 }
 
 // Listen starts listening.
-func (l *listener) Listen(ctx session.Context) {
+func (l *listener) Listen(ctx core.Context) {
 	ctx.Logger = ctx.Logger.With().
-		Str(session.LoggerKeyComponent, "listener").
+		Str(core.LoggerKeyComponent, "listener").
 		Logger()
 
 	for _, i := range l.Capture.Interfaces {
@@ -62,7 +62,7 @@ func (l *listener) Listen(ctx session.Context) {
 
 type activateFn func(inactive *pcap.InactiveHandle) (*pcap.Handle, error)
 
-func (l *listener) newHandle(ctx session.Context, i pcap.Interface, activate activateFn) (*pcap.Handle, error) {
+func (l *listener) newHandle(ctx core.Context, i pcap.Interface, activate activateFn) (*pcap.Handle, error) {
 	ctx.Logger = ctx.Logger.With().
 		Interface("interface", i).
 		Logger()
@@ -98,7 +98,7 @@ func (l *listener) newHandle(ctx session.Context, i pcap.Interface, activate act
 	return handle, nil
 }
 
-func (l *listener) readPackets(ctx session.Context, dataSource gopacket.PacketDataSource, decoder gopacket.Decoder) {
+func (l *listener) readPackets(ctx core.Context, dataSource gopacket.PacketDataSource, decoder gopacket.Decoder) {
 	source := gopacket.NewPacketSource(dataSource, decoder)
 	source.Lazy = true
 	source.NoCopy = true

capture/listener_test.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/pcap"
-	"github.com/rename-this/vhs/session"
+	"github.com/rename-this/vhs/core"
 	"gotest.tools/v3/assert"
 )
 
@@ -36,7 +36,7 @@ func TestNewHandle(t *testing.T) {
 	}
 	for _, c := range cases {
 		t.Run(c.desc, func(t *testing.T) {
-			ctx := session.NewContexts(&session.Config{}, &session.FlowConfig{}, nil)
+			ctx := core.NewContext(&core.Config{}, &core.FlowConfig{}, nil)
 			l := NewListener(&Capture{})
 			_, err := l.(*listener).newHandle(ctx, c.i, c.activate)
 			if c.errContains == "" {
@@ -107,7 +107,7 @@ func TestReadPackets(t *testing.T) {
 		t.Run(c.desc, func(t *testing.T) {
 			packets := c.listener.Packets()
 
-			ctx := session.NewContexts(&session.Config{DebugPackets: true}, &session.FlowConfig{}, nil)
+			ctx := core.NewContext(&core.Config{DebugPackets: true}, &core.FlowConfig{}, nil)
 
 			go c.listener.(*listener).readPackets(ctx, c.source, c.decoder)
 			for _, d := range c.source.data {

cmd/vhs/main.go

@@ -15,17 +15,16 @@ import (
 	"time"
 
 	"github.com/rename-this/vhs/capture"
+	"github.com/rename-this/vhs/core"
 	"github.com/rename-this/vhs/file"
 	"github.com/rename-this/vhs/flow"
 	"github.com/rename-this/vhs/gcs"
 	"github.com/rename-this/vhs/gzipx"
 	"github.com/rename-this/vhs/httpx"
 	"github.com/rename-this/vhs/internal/ioutilx"
 	"github.com/rename-this/vhs/jsonx"
-	"github.com/rename-this/vhs/middleware"
 	"github.com/rename-this/vhs/plugin"
 	"github.com/rename-this/vhs/s3compat"
-	"github.com/rename-this/vhs/session"
 	"github.com/rename-this/vhs/tcp"
 
 	_ "net/http/pprof"
@@ -48,8 +47,8 @@ func newRootCmd() *cobra.Command {
 			Short: "A tool for capturing and recording network traffic.",
 		}
 
-		cfg         = &session.Config{}
-		flowCfg     = &session.FlowConfig{}
+		cfg         = &core.Config{}
+		flowCfg     = &core.FlowConfig{}
 		inputLine   string
 		outputLines []string
 	)
@@ -99,10 +98,10 @@ func newRootCmd() *cobra.Command {
 	return cmd
 }
 
-func root(cfg *session.Config, flowCfg *session.FlowConfig, inputLine string, outputLines []string, parser *flow.Parser, logWriter io.Writer) error {
+func root(cfg *core.Config, flowCfg *core.FlowConfig, inputLine string, outputLines []string, parser *flow.Parser, logWriter io.Writer) error {
 	var (
 		errs = make(chan error, errBufSize)
-		ctx  = session.NewContextsForWriter(cfg, flowCfg, errs, logWriter)
+		ctx  = core.NewContextForWriter(cfg, flowCfg, errs, logWriter)
 	)
 
 	go func() {
@@ -210,13 +209,13 @@ func root(cfg *session.Config, flowCfg *session.FlowConfig, inputLine string, ou
 	return nil
 }
 
-func startMiddleware(ctx session.Context) (middleware.Middleware, error) {
+func startMiddleware(ctx core.Context) (core.Middleware, error) {
 	if ctx.FlowConfig.Middleware == "" {
 		ctx.Logger.Debug().Msg("no middleware configured")
 		return nil, nil
 	}
 
-	m, err := middleware.New(ctx, ctx.FlowConfig.Middleware)
+	m, err := core.NewMiddleware(ctx, ctx.FlowConfig.Middleware)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create middleware: %w", err)
 	}
@@ -259,10 +258,10 @@ func defaultParser() *flow.Parser {
 
 	p.LoadSink("gcs", gcs.NewSink)
 	p.LoadSink("s3compat", s3compat.NewSink)
-	p.LoadSink("stdout", func(_ session.Context) (flow.Sink, error) {
+	p.LoadSink("stdout", func(_ core.Context) (core.Sink, error) {
 		return os.Stdout, nil
 	})
-	p.LoadSink("discard", func(_ session.Context) (flow.Sink, error) {
+	p.LoadSink("discard", func(_ core.Context) (core.Sink, error) {
 		return ioutilx.NopWriteCloser(ioutil.Discard), nil
 	})
 	p.LoadSink("tcp", tcp.NewSink)

cmd/vhs/main_test.go

@@ -66,7 +66,7 @@ func TestRoot(t *testing.T) {
 					}
 				}()
 
-				return &session.FlowConfig{
+				return &core.FlowConfig{
 
 					SourceDuration:  10 * time.Second,
 					DrainDuration:   5 * time.Second,
@@ -114,7 +114,7 @@ func TestRoot(t *testing.T) {
 					}
 				}()
 
-				return &session.FlowConfig{
+				return &core.FlowConfig{
 					SourceDuration:  10 * time.Second,
 					DrainDuration:   5 * time.Second,
 					Addr:            strings.TrimLeft(s.URL, "http://"),
@@ -153,7 +153,7 @@ func TestRoot(t *testing.T) {
 		{
 			desc: "missing middleware",
 			init: func(ctx context.Context) *session.FlowConfig {
-				return &session.FlowConfig{
+				return &core.FlowConfig{
 					Middleware: "../../testdata/no_such_file",
 				}
 			},
@@ -162,7 +162,7 @@ func TestRoot(t *testing.T) {
 		{
 			desc: "middleware crash immediately",
 			init: func(ctx context.Context) *session.FlowConfig {
-				return &session.FlowConfig{
+				return &core.FlowConfig{
 					SourceDuration: time.Second,
 					Middleware:     "../../testdata/crash_immediately.bash",
 				}
@@ -173,7 +173,7 @@ func TestRoot(t *testing.T) {
 		{
 			desc: "middleware crash eventually",
 			init: func(ctx context.Context) *session.FlowConfig {
-				return &session.FlowConfig{
+				return &core.FlowConfig{
 					SourceDuration: 2 * time.Second,
 					Middleware:     "../../testdata/crash_eventually.bash",
 				}
@@ -184,7 +184,7 @@ func TestRoot(t *testing.T) {
 		{
 			desc: "bad input line",
 			init: func(ctx context.Context) *session.FlowConfig {
-				return &session.FlowConfig{}
+				return &core.FlowConfig{}
 			},
 			inputLine:             "---",
 			initializeErrContains: "invalid source: ---",
@@ -201,13 +201,13 @@ func TestRoot(t *testing.T) {
 				buf     bytes.Buffer
 			)
 
-			parser.Sinks["testout"] = func(session.Context) (flow.Sink, error) {
+			parser.Sinks["testout"] = func(core.Context) (flow.Sink, error) {
 				return ioutilx.NopWriteCloser(&buf), nil
 			}
 
 			var (
 				logBuf bytes.Buffer
-				cfg    = &session.Config{
+				cfg    = &core.Config{
 					ProfileHTTPAddr:   ":81112",
 					PrometheusAddr:    ":8080",
 					ProfilePathCPU:    "/tmp/vhs_cpu_test.prof",

core/close.go

@@ -0,0 +1,15 @@
+package core
+
+import "io"
+
+// CloseSequentially closes a and if there is no
+// error, closes b.
+func CloseSequentially(a, b io.Closer) error {
+	if err := a.Close(); err != nil {
+		return err
+	}
+	if b != nil {
+		return b.Close()
+	}
+	return nil
+}

core/close_test.go

@@ -0,0 +1,92 @@
+package core
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/rename-this/vhs/internal/ioutilx"
+	"gotest.tools/assert"
+)
+
+type TestDoubleOutputModifier struct {
+	OptCloseErr error
+}
+
+// Wrap wraps.
+func (m *TestDoubleOutputModifier) Wrap(w OutputWriter) (OutputWriter, error) {
+	tdom := &testDoubleOutputModifier{w: w}
+	if m.OptCloseErr == nil {
+		return tdom, nil
+	}
+
+	return &errWriteCloser{
+		Writer: tdom,
+		err:    m.OptCloseErr,
+	}, nil
+}
+
+type testDoubleOutputModifier struct {
+	w io.WriteCloser
+}
+
+func (o *testDoubleOutputModifier) Write(p []byte) (int, error) {
+	return o.w.Write(append(p, p...))
+}
+
+func (o *testDoubleOutputModifier) Close() error {
+	return o.w.Close()
+}
+
+type errWriteCloser struct {
+	io.Writer
+	err error
+}
+
+func (n *errWriteCloser) Close() error { return n.err }
+
+func TestCloseSequentially(t *testing.T) {
+	cases := []struct {
+		desc        string
+		a           OutputModifier
+		b           OutputModifier
+		errContains string
+	}{
+		{
+			desc: "no error",
+			a:    &TestDoubleOutputModifier{},
+			b:    &TestDoubleOutputModifier{},
+		},
+		{
+			desc:        "error a",
+			a:           &TestDoubleOutputModifier{OptCloseErr: errors.New("111")},
+			b:           &TestDoubleOutputModifier{},
+			errContains: "111",
+		},
+		{
+			desc:        "error b",
+			a:           &TestDoubleOutputModifier{},
+			b:           &TestDoubleOutputModifier{OptCloseErr: errors.New("222")},
+			errContains: "222",
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.desc, func(t *testing.T) {
+			var bufA bytes.Buffer
+			a, err := c.a.Wrap(ioutilx.NopWriteCloser(&bufA))
+			assert.NilError(t, err)
+
+			var bufB bytes.Buffer
+			b, err := c.b.Wrap(ioutilx.NopWriteCloser(&bufB))
+			assert.NilError(t, err)
+
+			err = CloseSequentially(a, b)
+			if c.errContains == "" {
+				assert.NilError(t, err)
+			} else {
+				assert.ErrorContains(t, err, c.errContains)
+			}
+		})
+	}
+}

session/config.go core/config.go

@@ -1,4 +1,4 @@
-package session
+package core
 
 import "time"
 

session/context.go core/context.go

@@ -1,4 +1,4 @@
-package session
+package core
 
 import (
 	"context"
@@ -15,14 +15,14 @@ const (
 	LoggerKeyComponent = "component"
 )
 
-// NewContexts creates a new set of contexts.
-func NewContexts(cfg *Config, flowCfg *FlowConfig, errs chan error) Context {
-	return NewContextsForWriter(cfg, flowCfg, errs, os.Stderr)
+// NewContext creates a new context.
+func NewContext(cfg *Config, flowCfg *FlowConfig, errs chan error) Context {
+	return NewContextForWriter(cfg, flowCfg, errs, os.Stderr)
 }
 
-// NewContextsForWriter creates a new set of contexts
+// NewContextForWriter creates a new context
 // with logs written to a specific writer.
-func NewContextsForWriter(cfg *Config, flowCfg *FlowConfig, errs chan error, w io.Writer) Context {
+func NewContextForWriter(cfg *Config, flowCfg *FlowConfig, errs chan error, w io.Writer) Context {
 	var (
 		sessionID      = ksuid.New().String()
 		registry       = envelope.NewRegistry()

session/context_test.go core/context_test.go

@@ -1,4 +1,4 @@
-package session
+package core
 
 import (
 	"errors"
@@ -12,7 +12,7 @@ func TestContexts(t *testing.T) {
 	var (
 		canceled = make(chan struct{}, 3)
 		errs     = make(chan error, 3)
-		ctx1     = NewContexts(&Config{}, &FlowConfig{}, errs)
+		ctx      = NewContext(&Config{}, &FlowConfig{}, errs)
 	)
 
 	fn := func(ctx Context) {
@@ -28,9 +28,9 @@ func TestContexts(t *testing.T) {
 		}
 	}
 
-	go fn(ctx1)
+	go fn(ctx)
 
-	ctx1.Cancel()
+	ctx.Cancel()
 
 	time.Sleep(500 * time.Millisecond)