transportproto

Overview ¶

Package transportproto is the line protocol between the emulator and a transport component.

Why there is a protocol at all ¶

The emulator delivers a domain event to whatever a subscription's target names. It knows three target types; the community will want others — Kafka, NATS, SQS, RabbitMQ — and none of those belong in a binary everybody downloads.

So a transport is a separate process, and this is what the two say to each other: newline-delimited JSON, requests on the child's stdin and responses on its stdout. It maps one-to-one onto the interface the emulator already had internally, which is what makes an out-of-process transport a drop-in for a compiled-in one rather than a second way of doing the same thing.

stdout is the protocol, stderr is the log ¶

The same split fft itself keeps. A transport writes diagnostics to stderr, which the emulator forwards to its own; anything it writes to stdout must be a frame. A component that prints a friendly startup banner on stdout breaks the protocol, and that is the right trade: one stream with one meaning is what makes the whole thing debuggable by reading it.

Frames ¶

→ {"id":1,"op":"hello"}
← {"id":1,"ok":true,"targets":["GOOGLE_CLOUD_PUB_SUB"],"status":"publishing to the Pub/Sub emulator at localhost:8085"}

→ {"id":2,"op":"plan","target":{"type":"GOOGLE_CLOUD_PUB_SUB","projectId":"p","topicId":"orders"}}
← {"id":2,"ok":true,"label":"p/orders"}

→ {"id":3,"op":"send","target":{…},"event":"ORDER_CREATED","data":{…}}
← {"id":3,"ok":true}

A refusal is ok:false with a reason, which is what the emulator logs when it skips a subscription. It is an ordinary answer, not an error: a target this transport cannot resolve is a subscription that will not fire, and the user needs to be told which and why.

Every request carries the target, because the child holds no state between frames. That costs a few bytes and removes handle lifetimes, reconnection and leak questions from a protocol whose whole job is to be obviously correct.

Compatibility ¶

Request, Response, Handler and Serve — together with MaxFrame and the OpHello/OpPlan/OpSend constants — are the public surface an external transport builds against, so their shape is a commitment. Version numbers that contract: a change that alters what an existing frame or field means bumps it, while a new optional field an older component can ignore does not.

The emulator hands the child the version it speaks in EnvVersion. fft does not enforce a match — it only reports it — so gating on it is the component's own choice: read EnvVersion, compare it against Version, and refuse a host you do not understand rather than fail in some more interesting way further in.

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"strconv"
	"syscall"

	"github.com/Joessst-Dev/fft-cli/pkg/transportproto"
)

// myTransport is a minimal [transportproto.Handler]. A real one would hold a broker
// client and resolve targets against it; this one answers with constants so the example
// stays about the wiring, not the broker. It is named and shaped to match the copy in
// the emulator component guide, so the two read as one example.
type myTransport struct{}

// Hello reports the target types this transport delivers, and one line for the
// emulator's startup notice saying where.
func (t *myTransport) Hello() (targets []string, status string, err error) {
	return []string{"MY_BROKER"}, "publishing to my-broker at localhost:9000", nil
}

// Plan resolves one subscription target into the label its deliveries are reported
// under, or returns an error the emulator logs as the reason it skips that subscription.
func (t *myTransport) Plan(target map[string]any) (label string, err error) {
	return "my-broker/orders", nil
}

// Send delivers one event to a target Plan accepted.
func (t *myTransport) Send(ctx context.Context, target map[string]any, event string, data []byte) error {
	return nil
}

// This is the whole of a transport component: refuse an emulator speaking a protocol
// version this build does not, then hand a [transportproto.Handler] to
// [transportproto.Serve], which owns the read/answer/write loop until stdin closes.
func main() {
	if got := os.Getenv(transportproto.EnvVersion); got != "" && got != strconv.Itoa(transportproto.Version) {
		fmt.Fprintf(os.Stderr, "the emulator speaks transport protocol %s, this component speaks %d\n", got, transportproto.Version)
		os.Exit(1)
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if err := transportproto.Serve(ctx, os.Stdin, os.Stdout, &myTransport{}); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

Output: