mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
zeekayandHanzo Dev
679ae61163
The engine took a PORT and the SDK an Options.HostPort, so neither end
could use a unix socket for traffic that never leaves the host. The
field name was not the only obstacle: luxfi/zap bound net.Listen("tcp",
":port") and reached a peer with net.DialTimeout("tcp", addr), so a path
was refused at the dialer with "missing port in address" and could not
be bound at all. That is fixed upstream (luxfi/zap Address + Network);
this pins the commit until it is tagged.
EmbedConfig.Address, Embedded.Address, ServeGated(addr) and
client.Options.Address each take ONE address, because a port is a detail
of one transport rather than a thing every listener has. A path binds or
dials a unix socket; anything else is host:port.
TestDialRequiresHostPort became TestDialRequiresAddress: Dial needs an
address, and asserts a socket path is refused for being absent rather
than for being portless. TestE2E_UnixSocket runs a real workflow over a
socket end to end, and TestE2E_UnixSocketRebind proves the socket is
unlinked on stop so a restart binds cleanly.
Co-authored-by: Hanzo Dev <[email protected]>
122 lines
3.1 KiB
Go
122 lines
3.1 KiB
Go
// Copyright © 2026 Hanzo AI. MIT License.
|
|
|
|
package examples
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/hanzoai/tasks/pkg/sdk/client"
|
|
tasks "github.com/hanzoai/tasks/pkg/tasks"
|
|
)
|
|
|
|
// RealtimeResult captures what arrived on the SSE stream during the
|
|
// demo window: every workflow.* event for the workflows we started.
|
|
type RealtimeResult struct {
|
|
StartedCount int
|
|
EventsCaught int
|
|
StartEvents int
|
|
CancelEvents int
|
|
}
|
|
|
|
// Realtime boots an embedded daemon, opens an SSE subscription via
|
|
// the native pkg/sdk/client.SubscribeEvents helper, fires a few
|
|
// state-changing operations, and tallies the events it observed
|
|
// before the context expires.
|
|
//
|
|
// This proves the realtime pipeline end-to-end: engine → broker →
|
|
// SSE → SDK channel → caller. No proxy, no gRPC, no upstream client.
|
|
func Realtime(parent context.Context) (*RealtimeResult, error) {
|
|
// Free port for the embedded daemon's HTTP+ZAP listeners.
|
|
l, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
port := l.Addr().(*net.TCPAddr).Port
|
|
_ = l.Close()
|
|
|
|
// pkg/tasks.Embed exposes a single ZAP listener; the HTTP shim is
|
|
// owned by cmd/tasksd. For the smoke test we want both, so we run
|
|
// a slim HTTP server with srv.HTTPHandler() + srv.EventsHandler().
|
|
bootCtx, bootCancel := context.WithCancel(parent)
|
|
defer bootCancel()
|
|
srv, err := tasks.Embed(bootCtx, tasks.EmbedConfig{
|
|
Address: fmt.Sprintf(":%d", port),
|
|
Namespace: "default",
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { _ = srv.Stop(context.Background()) }()
|
|
|
|
httpL, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
httpAddr := httpL.Addr().String()
|
|
go func() {
|
|
mux := http.NewServeMux()
|
|
mux.Handle("/v1/tasks/events", srv.EventsHandler())
|
|
mux.Handle("/v1/tasks/", srv.HTTPHandler())
|
|
_ = http.Serve(httpL, mux)
|
|
}()
|
|
|
|
// Subscribe BEFORE producing events so we see them all.
|
|
subCtx, subCancel := context.WithTimeout(parent, 2*time.Second)
|
|
defer subCancel()
|
|
ch, err := client.SubscribeEvents(subCtx, "http://"+httpAddr)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("subscribe: %w", err)
|
|
}
|
|
|
|
// Dial via ZAP and fire a couple state changes.
|
|
c, err := client.Dial(client.Options{
|
|
Address: fmt.Sprintf("127.0.0.1:%d", port),
|
|
Namespace: "default",
|
|
DialTimeout: 1 * time.Second,
|
|
CallTimeout: 1 * time.Second,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer c.Close()
|
|
|
|
res := &RealtimeResult{}
|
|
for i := 0; i < 3; i++ {
|
|
run, err := c.ExecuteWorkflow(subCtx, client.StartWorkflowOptions{
|
|
ID: fmt.Sprintf("realtime-%d-%d", port, i),
|
|
TaskQueue: "default",
|
|
}, "Realtime")
|
|
if err != nil {
|
|
return res, fmt.Errorf("execute %d: %w", i, err)
|
|
}
|
|
res.StartedCount++
|
|
// Cancel even-indexed ones so we exercise multiple event kinds.
|
|
if i%2 == 0 {
|
|
_ = c.CancelWorkflow(subCtx, run.GetID(), run.GetRunID())
|
|
}
|
|
}
|
|
|
|
// Drain events up to the context deadline.
|
|
for {
|
|
select {
|
|
case <-subCtx.Done():
|
|
return res, nil
|
|
case ev, ok := <-ch:
|
|
if !ok {
|
|
return res, nil
|
|
}
|
|
res.EventsCaught++
|
|
switch ev.Kind {
|
|
case "workflow.started":
|
|
res.StartEvents++
|
|
case "workflow.canceled":
|
|
res.CancelEvents++
|
|
}
|
|
}
|
|
}
|
|
}
|