bwilliams
6e101d2b4d
internal/events decouples the parts of JobSearch that cause events from
the parts that react to them. bus.go has Bus: Subscribe, and a synchronous
Publish that calls each handler in turn on the publisher's goroutine, so
handlers mustn't block (the hub doesn't: it queues per client). events.go
has the event types and their payloads, starting with OpportunitiesUpdated.
With fetcher.WithEvents, a save that stores new opportunities, or new
releases of stored ones, publishes OpportunitiesUpdated with their UUIDs and
counts. An opportunity whose row is only rewritten, as when it is fetched
again from the newest stored release, isn't counted as updated, and nothing
is published when nothing changed. store.SaveResult now reports
NewOpportunity for each opportunity saved.
job-search serve subscribes its WebSocket hub to the bus, so every client
gets {"type":"event","eventType":"OpportunitiesUpdated","payload":{"new":
[...],"updated":[...],"summary":{"new":n,"updated":n}}}. The Envelope gains
an eventType field, and Hub.PushEvent and Server.PushEvent send events.
79 lines
2.6 KiB
Go
79 lines
2.6 KiB
Go
// Package websocket implements the hub that pushes messages to connected
|
|
// clients: WebSocket clients upgraded by Hub.ServeHTTP (the /ws endpoint) and
|
|
// plain TCP clients run by Hub.ServeTCP, which send the same messages as
|
|
// newline-delimited JSON (see tcp.go). Package tcpserver runs a TCP server
|
|
// for those clients.
|
|
//
|
|
// Every message is a JSON Envelope: a WebSocket text frame, or one line on
|
|
// TCP.
|
|
//
|
|
// Client -> server:
|
|
//
|
|
// {"type":"ping","id":"1"}
|
|
// {"type":"echo","id":"2","payload":{...}}
|
|
//
|
|
// Server -> client:
|
|
//
|
|
// {"type":"welcome","payload":{"clientId":"..."}}
|
|
// {"type":"pong","id":"1","payload":{"time":"..."}}
|
|
// {"type":"echo","id":"2","payload":{...}}
|
|
// {"type":"error","id":"3","error":"..."}
|
|
// {"type":"item.created","payload":{...}}
|
|
// {"type":"event","eventType":"OpportunitiesUpdated","payload":{...}}
|
|
//
|
|
// Any other type from a client gets an "unknown message type" error. The
|
|
// last two are sent by Hub.Push, with the pushed message type, and by
|
|
// Hub.PushEvent, with TypeEvent and the event's type (see package events).
|
|
package websocket
|
|
|
|
import "encoding/json"
|
|
|
|
// Envelope is the wire format for every message in both directions.
|
|
// EventType is set only on a TypeEvent message, naming the event.
|
|
type Envelope struct {
|
|
Type string `json:"type"`
|
|
EventType string `json:"eventType,omitempty"`
|
|
ID string `json:"id,omitempty"`
|
|
Payload json.RawMessage `json:"payload,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
// Envelope types. TypeRequest and TypeAck are reserved: nothing sends or
|
|
// handles them yet.
|
|
const (
|
|
// TypeWelcome is sent to a client when it connects, with its client ID.
|
|
TypeWelcome = "welcome"
|
|
// TypePing asks the server for a TypePong.
|
|
TypePing = "ping"
|
|
// TypePong answers a TypePing, with the server's time.
|
|
TypePong = "pong"
|
|
// TypeEcho is sent back to the client with the same ID and payload.
|
|
TypeEcho = "echo"
|
|
// TypeEvent is an event pushed to clients by Hub.PushEvent, named by the
|
|
// envelope's EventType.
|
|
TypeEvent = "event"
|
|
// TypeRequest is reserved for requests from clients.
|
|
TypeRequest = "request"
|
|
// TypeAck is reserved for acknowledging a request.
|
|
TypeAck = "ack"
|
|
// TypeError reports an error, with the ID of the message it answers.
|
|
TypeError = "error"
|
|
)
|
|
|
|
// rawJSON returns value as raw JSON: value itself if it is already
|
|
// json.RawMessage, nil if value is nil or can't be marshalled.
|
|
func rawJSON(value any) json.RawMessage {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
if raw, ok := value.(json.RawMessage); ok {
|
|
return raw
|
|
}
|
|
|
|
encoded, err := json.Marshal(value)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return encoded
|
|
}
|