transport

package
v1.0.54 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 21, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package transport implements the cross-platform local IPC between the bus daemon and consume clients. The wire is \n-delimited JSON frames over a Unix domain socket (darwin/linux) or Windows Named Pipe; the abstraction keeps both endpoints (bus side / consume side) and both platforms behind one Listener/Dialer interface.

Frame protocol (see plan §9): 

consume → bus    hello, bye(reason=client_done), heartbeat, status_req
bus → consume    hello_ack, event, source_state, bye(reason=shutdown),
                 heartbeat, status_resp

Frames are independent JSON objects; no length prefix, no checksum. Maximum single-frame size is enforced by the Reader to bound memory and prevent a buggy/malicious peer from causing OOM (MaxFrameBytes).

Index

Constants

View Source 
const MaxFrameBytes = 1 << 20 // 1 MiB

MaxFrameBytes caps a single frame's wire size. 1 MiB accommodates large event payloads (kanban cards, doc snapshots) with headroom while still preventing a buggy/malicious peer from causing OOM via an unbounded line. Callers writing frames larger than this should split their work, not bump the cap — the SDK enforces its own cloud-side cap that is well below this.

Variables

View Source 
var ErrFrameTooLarge = errors.New("transport: frame exceeds MaxFrameBytes")

ErrFrameTooLarge is returned by Reader.Read when an incoming frame exceeds MaxFrameBytes. The underlying connection is left in an indeterminate state (the rest of the oversized frame is NOT drained) — callers should close the connection on receipt of this error.

Functions

func Dial 

func Dial(path string) (net.Conn, error)

Dial connects to an IPC endpoint at path. Returns a standard net.Conn the caller can wrap with NewReader/NewWriter. Closes are independent — closing the conn does not close the listener.

Types

type Bye 

type Bye struct {
	Type   FrameType `json:"type"`
	Reason string    `json:"reason"`
}

Bye is sent by either side at graceful shutdown. Reason is free-form for logs; structured shutdown causes are encoded in the value: 

"client_done"   — consume reached --max-events/--duration or SIGINT
"shutdown"      — bus SIGTERM/SIGINT
"idle_timeout"  — bus IdleTimeout fired with no consumers
"stop_request"  — bus received explicit Stop RPC

type Counters 

type Counters struct {
	Received uint64 `json:"received"`
	Dropped  uint64 `json:"dropped"`
}

Counters is the shared "received / dropped" pair used by per-consumer and per-event-type rollups.

type Event 

type Event struct {
	Type              FrameType         `json:"type"`
	Seq               uint64            `json:"seq"` // per-consumer monotonic, restarts at 1 on reconnect
	EventID           string            `json:"event_id"`
	EventBornTime     int64             `json:"event_born_time"`
	EventCorpID       string            `json:"event_corp_id,omitempty"`
	EventType         string            `json:"event_type"`
	EventUnifiedAppID string            `json:"event_unified_app_id,omitempty"`
	EventScope        string            `json:"event_scope,omitempty"`
	SubscribeID       string            `json:"subscribe_id,omitempty"`
	SourceID          string            `json:"source_id,omitempty"`
	RuleType          string            `json:"rule_type,omitempty"`
	Data              string            `json:"data"`
	Headers           map[string]string `json:"headers,omitempty"`
	ReceivedAtUnixMS  int64             `json:"received_at_unix_ms"`
}

Event wraps one delivered RawEvent for the wire. We keep the payload as a string (not nested JSON) so the bus does not need to parse / re-encode.

type FrameType 

type FrameType string

FrameType discriminates the JSON frames flowing over the IPC channel. Wire values are stable strings — bumping any of these is a protocol breaking change requiring a coordinated bus + consume rollout. 

const (
	FrameTypeHello       FrameType = "hello"        // consume → bus
	FrameTypeHelloAck    FrameType = "hello_ack"    // bus → consume
	FrameTypeEvent       FrameType = "event"        // bus → consume
	FrameTypeHeartbeat   FrameType = "heartbeat"    // bidirectional
	FrameTypeSourceState FrameType = "source_state" // bus → consume
	FrameTypeBye         FrameType = "bye"          // bidirectional
	FrameTypeStatusReq   FrameType = "status_req"   // consume/ad-hoc → bus
	FrameTypeStatusResp  FrameType = "status_resp"  // bus → consume/ad-hoc
)

func PeekType 

func PeekType(raw []byte) (FrameType, error)

PeekType returns the FrameType of a raw frame without fully decoding it. Returns an error when the JSON is malformed or the type field is missing.

type Heartbeat 

type Heartbeat struct {
	Type FrameType `json:"type"`
}

Heartbeat is bidirectional and stateless. It exists only to give both endpoints a chance to notice a dead peer via Read failure.

type Hello 

type Hello struct {
	Type        FrameType `json:"type"`
	ConsumerPID int       `json:"consumer_pid"`
	EventTypes  []string  `json:"event_types,omitempty"`  // wildcard list ("im.*", "approval.*"); empty = catch-all
	Filter      string    `json:"filter,omitempty"`       // optional regex over event_type
	SubscribeID string    `json:"subscribe_id,omitempty"` // optional personal subscription isolation key; empty = no subscribe_id filter
	Compact     bool      `json:"compact,omitempty"`      // hint to status output; bus does not transform payloads
	// Role distinguishes a real consumer (registered for events) from an
	// ad-hoc tooling connection (status/list/stop). Ad-hoc connections do
	// NOT register with the Hub.
	Role HelloRole `json:"role,omitempty"`
}

Hello is the first frame a consumer sends after dialing the bus. The bus uses event_types + filter to do server-side pushdown (plan §1 decision: only events matching this filter are written to this consumer's sendCh).

type HelloAck 

type HelloAck struct {
	Type               FrameType `json:"type"`
	BusPID             int       `json:"bus_pid"`
	SourceState        string    `json:"source_state"`                // mirrors source.State string value
	StateSource        string    `json:"state_source"`                // "hook" | "inferred"
	ClientIDSource     string    `json:"client_id_source"`            // auth.CredentialSource string
	ClientSecretSource string    `json:"client_secret_source"`        // auth.CredentialSource string
	IdleTimeoutSecs    int       `json:"idle_timeout_secs,omitempty"` // bus's IdleTimeout for diagnostics
}

HelloAck is the bus's reply on accepted Hello. SourceState/StateSource mirror the source.Machine snapshot at connect time; CredentialsSource fields tell users which channel actually supplied the credentials in use (env / app_config / keychain / plain_config), see plan "凭证来源拆字段".

type HelloRole 

type HelloRole string

HelloRole tags the purpose of a Hello connection. 

const (
	HelloRoleConsumer HelloRole = ""       // default
	HelloRoleStatus   HelloRole = "status" // event list / event status
	HelloRoleStop     HelloRole = "stop"   // event stop (graceful trigger)
)

type Listener 

type Listener interface {
	// Accept blocks until a peer connects or the listener is closed.
	Accept() (net.Conn, error)
	// Close unbinds the endpoint and unblocks pending Accept calls.
	Close() error
	// Endpoint returns the human-readable address (Unix path or Windows pipe
	// name) — only used in log messages and status output.
	Endpoint() string
}

Listener is the bus-side accept loop. Implementations bind a Unix socket (darwin/linux) or Windows Named Pipe at the given Endpoint; calling Close removes the underlying socket file on Unix.

func Listen 

func Listen(path string) (Listener, error)

Listen binds an IPC endpoint at path. On Unix path is a filesystem socket path (caller must ensure the parent directory exists with mode 0700). On Windows path is a Named Pipe name like `\\.\pipe\dws-event-<edition>-<hash>`.

Stale Unix sockets (left behind by a crashed bus) are unlinked automatically before bind. Caller MUST hold the bus.lock before calling Listen so this unlink is race-safe against a still-running bus.

type Reader 

type Reader struct {
	// contains filtered or unexported fields
}

Reader reads \n-delimited JSON frames from an underlying byte stream. Wrap each connection (one per direction) in a Reader; Reader is not safe for concurrent use.

func NewReader 

func NewReader(r io.Reader) *Reader

NewReader returns a Reader with a buffer sized to MaxFrameBytes so a single frame can fit in the buffer without growing.

func (*Reader) Read 

func (r *Reader) Read() ([]byte, error)

Read returns the next frame's raw bytes (the terminating \n is stripped). Returns io.EOF cleanly when the peer closes the connection between frames. Returns ErrFrameTooLarge when a single frame exceeds the cap.

func (*Reader) ReadJSON 

func (r *Reader) ReadJSON(dst any) error

ReadJSON decodes the next frame into dst. dst must be a non-nil pointer.

type SourceState 

type SourceState struct {
	Type        FrameType `json:"type"`
	State       string    `json:"state"`        // source.State string
	StateSource string    `json:"state_source"` // hook | inferred
	Attempt     int       `json:"attempt,omitempty"`
}

SourceState is pushed bus → consume whenever the connection state machine transitions to / from connected. Consumers may render it; v1 they just forward to stderr when not --quiet.

type StatusBus 

type StatusBus struct {
	PID            int                 `json:"pid"`
	UptimeSecs     int64               `json:"uptime_secs"`
	IdleTimeoutSec int                 `json:"idle_timeout_secs"`
	ClientID       string              `json:"client_id"`
	Edition        string              `json:"edition"`
	SourceKind     dwsevent.SourceKind `json:"source_kind,omitempty"`
	IdentityHash   string              `json:"identity_hash,omitempty"`
	SourceID       string              `json:"source_id,omitempty"`
}

StatusBus is the bus daemon's identity / lifecycle view.

type StatusConsumer 

type StatusConsumer struct {
	PID            int      `json:"pid"`
	EventTypes     []string `json:"event_types,omitempty"`
	Filter         string   `json:"filter,omitempty"`
	SubscribeID    string   `json:"subscribe_id,omitempty"`
	SubscribedAtMS int64    `json:"subscribed_at_ms"`
	Received       uint64   `json:"received"`
	Dropped        uint64   `json:"dropped"`
}

StatusConsumer is one consumer's per-IPC-connection view.

type StatusReq 

type StatusReq struct {
	Type FrameType `json:"type"`
}

StatusReq is an empty JSON frame ad-hoc tooling sends after Hello to request a full StatusResp. Bus replies with one StatusResp then closes the connection.

type StatusResp 

type StatusResp struct {
	Type                 FrameType           `json:"type"`
	Bus                  StatusBus           `json:"bus"`
	SourceState          StatusSource        `json:"source_state"`
	Consumers            []StatusConsumer    `json:"consumers"`
	PerEventTypeCounters map[string]Counters `json:"per_event_type"`
}

StatusResp is the bus's snapshot view for `dws event status` rendering. Counts are accumulated since bus start; per-consumer entries are sorted by PID for deterministic output.

type StatusSource 

type StatusSource struct {
	State           string `json:"state"`
	Source          string `json:"source"`
	LastEventAtMS   int64  `json:"last_event_at_ms,omitempty"`
	LastReconnectMS int64  `json:"last_reconnect_at_ms,omitempty"`
	ReconnectCount  int    `json:"reconnect_count"`
}

StatusSource is the source.Machine snapshot at status RPC time.

type Writer 

type Writer struct {
	// contains filtered or unexported fields
}

Writer marshals values to JSON and writes them as \n-terminated frames. Writer is not safe for concurrent use; callers wanting a fan-in writer must serialise externally.

func NewWriter 

func NewWriter(w io.Writer) *Writer

NewWriter returns a Writer wrapping w. The Writer does NOT buffer — every WriteJSON call performs a single Write to the underlying stream so backpressure is immediate.

func (*Writer) WriteJSON 

func (w *Writer) WriteJSON(v any) error

WriteJSON serialises v and writes it as one frame. Refuses to write frames larger than MaxFrameBytes with ErrFrameTooLarge.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.