G findings4 pipeline: the runner's views reader writes its findings through mesh-sdk's finding writer
Builds: design 47 G §689, §686 (A-3), §639, A-13 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -0,0 +1,33 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
)
|
||||
|
||||
// findingWriter is the runner's mesh-sdk finding writer (design 47 G §461, §689): the runner observes only what its views
|
||||
// reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer
|
||||
// forms its subjects itself (design 47 G §383, A-13).
|
||||
func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) {
|
||||
w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
|
||||
Module: cred.Module, State: state, Node: cred.Node, JS: js, Conn: conn, Logf: logf,
|
||||
Observe: func(context.Context) (views.Observed[condseat.Finding], error) {
|
||||
return views.Observed[condseat.Finding]{}, nil
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// A writer that cannot start (the stream of its asks is not there yet) is said; its findings are then kept in the
|
||||
// state alone and said by the next start (as the node-engine's is).
|
||||
if err := w.Start(ctx); err != nil {
|
||||
logf("the finding writer cannot start, its findings are kept in the state alone: %v", err)
|
||||
}
|
||||
return w, nil
|
||||
}
|
||||
@@ -56,7 +56,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
defer calls.Close()
|
||||
hub := process.NewEventHub(logger.Printf)
|
||||
rv, err := startViews(ctx, js, conn, cred, hub, calls.Call, logger.Printf)
|
||||
rv, err := startViews(ctx, js, conn, cred, hub, calls.State("findings"), calls.Call, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -5,10 +5,9 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
|
||||
sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings"
|
||||
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
|
||||
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -30,11 +29,14 @@ type runnerViews struct {
|
||||
|
||||
// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
|
||||
// after, then start hub once.
|
||||
func startViews(ctx context.Context, js jetstream.JetStream, conn process.Subscriber, cred process.Account, hub *process.EventHub,
|
||||
func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cred process.Account, hub *process.EventHub, findingsState process.State,
|
||||
call meshcatalogclient.Call, logf func(string, ...any)) (*runnerViews, error) {
|
||||
findings := process.AwaitBucket(ctx, js, conn, mcc.FindingsBucket(cred.Module), "a view waited for past five minutes is said in the log alone", logf)
|
||||
findings, err := findingWriter(ctx, cred, conn, js, findingsState, logf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
status := process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "the runner's first views are not said in its first-view tool, nor asked again by hand", logf)
|
||||
writer := func() views.FindingWriter { return sdkfindings.Writer(ctx, findings.KV) }
|
||||
writer := func() views.FindingWriter { return views.WriterFindings{Writer: findings} }
|
||||
reader := &views.ProcessReader{Module: cred.Module, Call: call, Logf: logf, Hub: hub, Findings: writer}
|
||||
|
||||
// A watched branch first named by a registration gets its view of the catalog half (CT10), its conditions and its
|
||||
|
||||
@@ -9,8 +9,8 @@ require (
|
||||
)
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000
|
||||
git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000
|
||||
|
||||
Reference in New Issue
Block a user