Documentation
¶
Overview ¶
Package source wraps the open-dingtalk Stream SDK and exposes a small blocking Start interface plus a connection state machine. The SDK is the only place in the bus that talks to the cloud; the rest of the bus stays vendor-agnostic and would be drop-in replaceable for a different Stream provider.
Index ¶
Constants ¶
const ( PortalTicketModeNormal = "normal" PortalTicketModeCustom = "custom" )
Variables ¶
var IdleAfter = 60 * time.Second
IdleAfter is the duration of no events after which StateConnected transitions to StateIdle. Exposed for tests to bypass real-time waits.
Functions ¶
This section is empty.
Types ¶
type AccessTokenProvider ¶ added in v1.0.54
type Config ¶
type Config struct {
ClientID string
ClientSecret string
// PortalTicket switches the source to the portal-managed user Stream
// ticket flow. When set, Start fetches endpoint+ticket over HTTP and
// dials the returned WebSocket directly instead of asking the SDK to
// open an app-credential connection.
PortalTicket *PortalTicketConfig
// Now is injected for tests. Defaults to time.Now when nil.
Now func() time.Time
}
Config carries the credentials and behavioural knobs needed to construct a DingtalkSource. ClientID/ClientSecret are required for app-credential SDK mode. Portal ticket normal mode uses portal-side managed credentials, so it does not require local app credentials.
type DingtalkSource ¶
type DingtalkSource struct {
// contains filtered or unexported fields
}
DingtalkSource is the cloud-side adapter. It owns one StreamClient and one state Machine; lifecycle is bounded by the context passed to Start.
func New ¶
func New(cfg Config, _ ...SourceOption) (*DingtalkSource, error)
New constructs a DingtalkSource. Returns an error if required Config fields are missing — keep the boundary tight so misconfiguration fails loudly rather than at first-event time.
func (*DingtalkSource) Start ¶
Start opens the Stream WebSocket and blocks until ctx is cancelled or a fatal SDK error occurs. Events are delivered to emit synchronously from the SDK callback goroutine — emit MUST return immediately (the SDK's processLoop is single-goroutine; see plan invariant #1 and P0 §1 row 2).
Blocking semantics: the underlying StreamClient.Start returns as soon as dial succeeds (it spawns its own processLoop goroutine), so we wait on ctx.Done() before returning. cli.Close() is always called on exit; ctx.Err is returned for context cancellation, otherwise the underlying error.
func (*DingtalkSource) State ¶
func (s *DingtalkSource) State() Snapshot
State returns the current Snapshot of the connection state machine.
type ForceRefreshTokenFn ¶ added in v1.0.54
ForceRefreshTokenFn rotates an access token that the server has just rejected (HTTP 401). It receives the exact rejected token so the caller's compare-and-refresh logic can skip the refresh when another goroutine has already rotated it, and returns the fresh token to retry with. Optional: when nil a 401 stays fatal, matching the previous behavior.
type Machine ¶
type Machine struct {
// contains filtered or unexported fields
}
Machine tracks the connection state. Methods are safe for concurrent use.
func NewMachine ¶
func NewMachine() *Machine
NewMachine returns a Machine in the StateDisconnected initial state.
func (*Machine) OnConnected ¶
func (m *Machine) OnConnected()
OnConnected moves to StateConnected. Called when SDK Start returns nil.
func (*Machine) OnConnecting ¶
func (m *Machine) OnConnecting()
OnConnecting moves to StateConnecting. Called by Source.Start before the SDK dial completes.
func (*Machine) OnEvent ¶
func (m *Machine) OnEvent()
OnEvent records the timestamp of the most recently received event. Resets any Reconnecting or Idle state back to Connected (the connection is demonstrably alive).
func (*Machine) OnReconnect ¶
func (m *Machine) OnReconnect()
OnReconnect marks a reconnect signal observed (from SDK log or future hook). Increments the counter and stamps the time.
func (*Machine) OnStopped ¶
func (m *Machine) OnStopped()
OnStopped marks the SDK as closed (graceful exit or Close()).
type PersonalConfig ¶
type PersonalConfig struct {
AccessToken string
AccessTokenProvider AccessTokenProvider
ForceRefreshToken ForceRefreshTokenFn
ClientID string
ClientSecret string
SourceID string
TicketURL string
TicketMode string
HTTPClient *http.Client
WebSocketDialer *websocket.Dialer
Now func() time.Time
ReconnectMin time.Duration
ReconnectMax time.Duration
}
type PersonalSource ¶
type PersonalSource struct {
// contains filtered or unexported fields
}
func NewPersonal ¶
func NewPersonal(cfg PersonalConfig) (*PersonalSource, error)
func (*PersonalSource) State ¶
func (s *PersonalSource) State() Snapshot
type PortalTicketConfig ¶
type PortalTicketConfig struct {
TicketURL string
AccessToken string
AccessTokenProvider AccessTokenProvider
ForceRefreshToken ForceRefreshTokenFn
SourceID string
Mode string
ClientID string
ClientSecret string
UserAgent string
HTTPClient *http.Client
WebSocketDialer *websocket.Dialer
ReconnectMin time.Duration
ReconnectMax time.Duration
DisableReconnect bool
}
PortalTicketConfig describes the portal-managed user Stream ticket flow. normal mode uses portal-side managed credentials; custom mode asks portal to open the user connection with the caller-provided clientId/clientSecret.
func (*PortalTicketConfig) Valid ¶
func (c *PortalTicketConfig) Valid() error
type Snapshot ¶
type Snapshot struct {
State State `json:"state"`
StateSource SourceKind `json:"state_source"`
LastEventAt time.Time `json:"last_event_at,omitempty"`
LastReconnectAt time.Time `json:"last_reconnect_at,omitempty"`
ReconnectCount int `json:"reconnect_count"`
}
Snapshot returns the full state view for status rendering: state + source + last event/reconnect timestamps + reconnect count.
type SourceKind ¶
type SourceKind string
SourceKind identifies whether the state was reported by an authoritative SDK hook or inferred from observable side-channels.
const ( // SourceHook is reserved for future SDK versions that expose // ConnectionState / keepalive callbacks. v1 does not emit this. SourceHook SourceKind = "hook" // SourceInferred is the only value emitted by v1. SourceInferred SourceKind = "inferred" )
type SourceOption ¶
type SourceOption func(*sourceConfig)
SourceOptions is the functional-option type reserved for future overlay extensions (e.g. inject trace IDs into emit, swap the underlying client for a fake). v1 takes no options; the type exists so callers can write `New(cfg)` today and `New(cfg, WithFoo())` tomorrow without an API break. See plan §7 Edition 扩展点.
type State ¶
type State string
State enumerates the connection states surfaced by `dws event status`.
Stream SDK v0.9.1 does not expose a user-registerable ConnectionState hook (verified P0 §1 escape-hatch table row #4), so the state machine runs in "inferred" mode: it derives state from observable signals — Start() return value, last-event-received timestamp, last-error timestamp, SDK reconnect log lines parsed by the source layer. The Source field `state_source` (env-style annotation) is emitted alongside the state so downstream tools can distinguish authoritative from inferred values when SDK hook support arrives in a future release.
const ( // StateDisconnected is the zero state before Start has been called. StateDisconnected State = "disconnected" // StateConnecting is set on entry to Start, before SDK dial completes. StateConnecting State = "connecting" // StateConnected is set when Start returns nil (dial succeeded). StateConnected State = "connected" // StateIdle is set when the connection is still up but no event has been // observed for a quiet window (default 60s). Distinguishes "everything // fine, just no traffic" from "looks connected but broken". StateIdle State = "idle" // StateReconnecting is set when the SDK has signalled a reconnect (via // log line parse or future hook). Cleared on next event received. StateReconnecting State = "reconnecting" // StateDegraded is set when reconnects keep happening but no events flow. // Useful to surface "looks alive, actually broken" cases via status. StateDegraded State = "degraded" // StateStopped is set after Close() / context cancellation. StateStopped State = "stopped" )
Source Files