sad

cmd/cgat/main.go

@@ -9,7 +9,9 @@ import (
 	"pggat2/lib/bouncer/bouncers/v0"
 	"pggat2/lib/bouncer/frontends/v0"
 	"pggat2/lib/middleware/middlewares/onebuffer"
-	"pggat2/lib/middleware/middlewares/unterminate"
+	"pggat2/lib/mw2"
+	"pggat2/lib/mw2/interceptor"
+	"pggat2/lib/mw2/middlewares/unterminate"
 	"pggat2/lib/rob"
 	"pggat2/lib/rob/schedulers/v2"
 	"pggat2/lib/zap"
@@ -59,9 +61,11 @@ func main() {
 		go func() {
 			source := r.NewSource()
 			client := zio.MakeReadWriter(conn)
-			ut := unterminate.MakeUnterminate(&client)
-			frontends.Accept(ut)
 			ob := onebuffer.MakeOnebuffer(&client)
+			mw := interceptor.MakeInterceptor(&ob, []mw2.Middleware{
+				unterminate.Unterminate,
+			})
+			frontends.Accept(&mw)
 			done := make(chan struct{})
 			defer close(done)
 			for {
@@ -70,7 +74,7 @@ func main() {
 					break
 				}
 				source.Schedule(job{
-					client: &ob,
+					client: &mw,
 					done:   done,
 				}, 0)
 				<-done

lib/middleware/middlewares/unterminate/unterminate.go

-package unterminate
-
-import (
-	"io"
-
-	"pggat2/lib/zap"
-	packets "pggat2/lib/zap/packets/v3.0"
-)
-
-type Unterminate struct {
-	zap.ReadWriter
-}
-
-func MakeUnterminate(inner zap.ReadWriter) Unterminate {
-	return Unterminate{
-		ReadWriter: inner,
-	}
-}
-
-func (T Unterminate) Read() (zap.In, error) {
-	in, err := T.ReadWriter.Read()
-	if err != nil {
-		return zap.In{}, err
-	}
-	if in.Type() == packets.Terminate {
-		return zap.In{}, io.EOF
-	}
-	return in, nil
-}
-
-var _ zap.ReadWriter = Unterminate{}

lib/mw2/interceptor/context.go

+package interceptor
+
+import (
+	"pggat2/lib/mw2"
+	"pggat2/lib/zap"
+)
+
+type Context struct {
+	cancelled bool
+	zap.ReadWriter
+}
+
+func makeContext(rw zap.ReadWriter) Context {
+	return Context{
+		ReadWriter: rw,
+	}
+}
+
+func (T *Context) reset() {
+	T.cancelled = false
+}
+
+func (T *Context) Cancel() {
+	T.cancelled = true
+}
+
+var _ mw2.Context = (*Context)(nil)

lib/mw2/interceptor/interceptor.go

+package interceptor
+
+import (
+	"pggat2/lib/mw2"
+	"pggat2/lib/util/decorator"
+	"pggat2/lib/zap"
+)
+
+type Interceptor struct {
+	noCopy decorator.NoCopy
+
+	middlewares []mw2.Middleware
+	Context
+}
+
+func MakeInterceptor(rw zap.ReadWriter, middlewares []mw2.Middleware) Interceptor {
+	return Interceptor{
+		middlewares: middlewares,
+		Context:     makeContext(rw),
+	}
+}
+
+func (T *Interceptor) Read() (zap.In, error) {
+	for {
+		in, err := T.ReadWriter.Read()
+		if err != nil {
+			return zap.In{}, err
+		}
+
+		T.Context.reset()
+		for _, mw := range T.middlewares {
+			err = mw.Read(&T.Context, in)
+			if err != nil {
+				return zap.In{}, err
+			}
+			if T.cancelled {
+				break
+			}
+		}
+
+		if !T.cancelled {
+			return in, nil
+		}
+	}
+}
+
+func (T *Interceptor) ReadUntyped() (zap.In, error) {
+	for {
+		in, err := T.ReadWriter.ReadUntyped()
+		if err != nil {
+			return zap.In{}, err
+		}
+
+		T.Context.reset()
+		for _, mw := range T.middlewares {
+			err = mw.Read(&T.Context, in)
+			if err != nil {
+				return zap.In{}, err
+			}
+			if T.cancelled {
+				break
+			}
+		}
+
+		if !T.cancelled {
+			return in, nil
+		}
+	}
+}
+
+func (T *Interceptor) Send(out zap.Out) error {
+	T.Context.reset()
+	for _, mw := range T.middlewares {
+		err := mw.Send(&T.Context, out)
+		if err != nil {
+			return err
+		}
+		if T.cancelled {
+			break
+		}
+	}
+
+	if !T.cancelled {
+		return T.Context.ReadWriter.Send(out)
+	}
+	return nil
+}
+
+var _ zap.ReadWriter = (*Interceptor)(nil)

lib/mw2/middlewares/unterminate/unterminate.go

+package unterminate
+
+import (
+	"io"
+
+	"pggat2/lib/mw2"
+	"pggat2/lib/zap"
+	packets "pggat2/lib/zap/packets/v3.0"
+)
+
+var Unterminate = unterm{}
+
+type unterm struct {
+	mw2.Nil
+}
+
+func (unterm) Read(_ mw2.Context, in zap.In) error {
+	if in.Type() == packets.Terminate {
+		return io.EOF
+	}
+	return nil
+}
+
+var _ mw2.Middleware = unterm{}