Documentation
¶
Overview ¶
Package stream-oee demonstrates how to govern OEE computation with forge and bridge it to a reactive machine event stream using the stream package.
The design pattern ¶
forge governs the WHAT — each OEE component is a separately signed forge.Function with a SHA-256 contract hash, Author, and ApprovedBy. Changing a formula invalidates its hash, requiring re-approval. The Registry produces a governance YAML.
stream governs the WHEN — machine events arrive as a push-based stream. stream.Window collects events over a time window; when the window fires, the governed forge functions run on the batch.
Key design decision: sequential wrapper vs Tee+CombineLatest3 ¶
When all three OEE components derive from the SAME event window, run them sequentially inside one wrapper forge function rather than using Tee+CombineLatest3. Tee+CombineLatest3 is correct when inputs arrive on INDEPENDENT streams (separate MQTT topics with different frequencies). For a single window feeding three functions, sequential application is cleaner, avoids sync issues, and the pipeline Registry still shows all six functions with their individual hashes.
Tapping into partial calculations ¶
To observe intermediate values (Availability, Performance, Quality), the wrapper forge function returns an OEEResult struct carrying all components alongside the final OEE. This is the idiomatic pattern for exposing intermediate values in a stream pipeline — returning a rich result type rather than splitting via Tee.
Downstream stream.Tap observers can then react independently to any field:
results = stream.Tap(ctx, results, func(r OEEResult) {
if float64(r.Availability) < 0.80 { maintenanceTeam.Notify() }
if float64(r.Quality) < 0.90 { qualityTeam.Notify() }
})
Architecture ¶
MachineEvent stream (goroutine / MQTT)
→ stream.Window(windowDuration) → Stream[[]MachineEvent]
→ stream.Apply(computeOEEFromWindow) → Stream[OEEResult]
(OEEResult{Availability, Performance, Quality, OEE})
→ stream.Tap: Availability < 80% → maintenance pre-alert
→ stream.Tap: Quality < 90% → quality team notification
→ stream.Tee ───────────────────────────────────────
↓ alert branch ↓ metrics branch
Filter(oee < target) FlatMapSlice → []ComponentMetric
Drain(alert) Drain(metrics bus)
Running ¶
go run ./examples/stream-oee
Source Files