fizeau/service_execute_dispatch.go at master · easel/fizeau · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
package fizeau

import (
	"context"
	"sync/atomic"
	"time"

	"github.com/easel/fizeau/internal/harnesses"
	"github.com/easel/fizeau/internal/serviceimpl"
)

type executeRouteResolver interface {
	resolveExecuteRoute(ServiceExecuteRequest) (*RouteDecision, error)
	resolveExecuteRouteContext(context.Context, ServiceExecuteRequest) (*RouteDecision, error)
}

type executeSessionLogOpener interface {
	openSessionLog(ServiceExecuteRequest, RouteDecision, string) *serviceSessionLog
}

type executeEventFanout interface {
	OpenSession(string)
	BroadcastEvent(string, ServiceEvent)
	CloseSession(string, ServiceEvent)
}

type executeRunnerInvoker interface {
	dispatchExecuteRun(context.Context, executeRunContext)
}

type executeRunContext struct {
	req      ServiceExecuteRequest
	decision RouteDecision
	meta     map[string]string
	out      chan<- ServiceEvent
	seq      *atomic.Int64
	start    time.Time
	sl       *serviceSessionLog
	session  string
}

func (s *service) executeRouteResolver() executeRouteResolver {
	return s
}

func (s *service) executeSessionLogOpener() executeSessionLogOpener {
	return s
}

func (s *service) executeEventFanout() executeEventFanout {
	return s.hub
}

func (s *service) executeRunnerInvoker() executeRunnerInvoker {
	return s
}

func (s *service) dispatchExecuteRun(ctx context.Context, run executeRunContext) {
	decision := serviceimpl.ExecuteRunnerDecision{
		Harness:  run.decision.Harness,
		Provider: run.decision.Provider,
		Model:    run.decision.Model,
	}
	serviceimpl.DispatchExecuteRun(ctx, serviceimpl.ExecuteDispatchRequest{
		Decision: decision,
		Started:  run.start,
	}, serviceimpl.ExecuteDispatchCallbacks{
		RunNative: func(ctx context.Context) {
			s.runNative(ctx, run.req, run.decision, run.meta, run.out, run.seq, run.start, run.sl, run.session)
		},
		RunSubprocess: func(ctx context.Context, runner harnesses.Harness) {
			s.runSubprocess(ctx, run.req, run.decision, run.meta, run.out, run.seq, run.start, run.sl, run.session, runner)
		},
		RunVirtual: func(ctx context.Context) {
			s.runVirtual(ctx, run.req, run.decision, run.meta, run.out, run.seq, run.start, run.sl, run.session)
		},
		RunScript: func(ctx context.Context) {
			s.runScript(ctx, run.req, run.decision, run.meta, run.out, run.seq, run.start, run.sl, run.session)
		},
		IsHTTPProvider: func(harness string) bool {
			cfg, ok := s.registry.Get(harness)
			return ok && cfg.IsHTTPProvider
		},
		Finalize: func(final harnesses.FinalData) {
			s.recordRouteAttemptFromFinal(final)
			finalizeAndEmit(run.out, run.seq, run.meta, run.req, run.sl, final)
		},
	})
}