Documentation
¶
Index ¶
- func DefaultCheckpointPath() (string, error)
- func NewGitHubClient(token, baseURL string) (*github.Client, error)
- type Checkpoint
- type CheckpointKey
- type CheckpointStore
- func (s *CheckpointStore) Delete(key CheckpointKey)
- func (s *CheckpointStore) Get(key CheckpointKey) (Checkpoint, bool)
- func (s *CheckpointStore) Set(key CheckpointKey, cp Checkpoint)
- func (s *CheckpointStore) Shutdown(ctx context.Context) error
- func (s *CheckpointStore) ShutdownTimeout(timeout time.Duration) error
- type Config
- type ConnectionInfo
- type DispatchItem
- type Dispatcher
- type Poller
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DefaultCheckpointPath ¶ added in v1.0.51
DefaultCheckpointPath returns the default checkpoint file path, ~/.config/plan42-runner.checkpoint.json.
func NewGitHubClient ¶ added in v1.0.54
NewGitHubClient builds a go-github client authenticated with the given token and targeting the given base URL. For public GitHub (or empty URL) it targets https://api.github.com; otherwise it configures enterprise URLs.
Types ¶
type Checkpoint ¶
type Checkpoint struct {
LastEventID string `json:"last_event_id"`
ETag string `json:"etag"`
PollIntervalSecs int `json:"poll_interval_seconds"`
}
Checkpoint holds the durable state for a single polling pair. The JSON tags define the on-disk file format.
type CheckpointKey ¶
CheckpointKey identifies a unique polling pair.
type CheckpointStore ¶
type CheckpointStore struct {
// contains filtered or unexported fields
}
CheckpointStore persists polling checkpoints to disk so a runner restart doesn't replay the entire Events API window. Writes are debounced: the in-memory map is the canonical state, and a timer-driven flush writes it to disk at most once per flushDebounce interval.
func NewCheckpointStore ¶
func NewCheckpointStore(path string, clk clock.Clock) (*CheckpointStore, error)
NewCheckpointStore creates a CheckpointStore backed by the file at path, using clk for its debounce timer. The parent directory is created if needed. Any existing file is loaded; a missing or unparseable file is treated as empty. The debounce timer (and the goroutine that watches it) is created lazily on the first mutation.
func (*CheckpointStore) Delete ¶
func (s *CheckpointStore) Delete(key CheckpointKey)
Delete removes the checkpoint for the given key and marks the store dirty.
func (*CheckpointStore) Get ¶
func (s *CheckpointStore) Get(key CheckpointKey) (Checkpoint, bool)
Get returns the checkpoint for the given key.
func (*CheckpointStore) Set ¶
func (s *CheckpointStore) Set(key CheckpointKey, cp Checkpoint)
Set updates the checkpoint for the given key and marks the store dirty.
func (*CheckpointStore) Shutdown ¶
func (s *CheckpointStore) Shutdown(ctx context.Context) error
Shutdown stops the checkpoint store: cancels the watch goroutine, waits for any in-flight timer-scheduled flush, and runs a synchronous final flush.
func (*CheckpointStore) ShutdownTimeout ¶
func (s *CheckpointStore) ShutdownTimeout(timeout time.Duration) error
ShutdownTimeout calls Shutdown with a timeout-bounded context.
type Config ¶ added in v1.0.51
type Config struct {
Checkpoints *CheckpointStore
// EventCapacity is the buffer size of the event channel. Default 200.
EventCapacity int
}
Config holds the dependencies for the github events poller.
type ConnectionInfo ¶
ConnectionInfo carries the connection details needed to construct a GitHub client for polling a (GithubConnectionID, OrgName) pair.
type DispatchItem ¶ added in v1.0.51
type DispatchItem struct {
// contains filtered or unexported fields
}
DispatchItem pairs an event with the github client a handler should use to act on it. The client is authenticated for the event's connection.
type Dispatcher ¶ added in v1.0.51
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher reads events from a channel and invokes HandlerRegistry.Handle via a fixed-size worker pool. It owns a single ContextGroup that tracks the worker goroutines and provides the context passed to handlers.
func NewDispatcher ¶ added in v1.0.51
func NewDispatcher(registry *handlers.HandlerRegistry, eventCh <-chan DispatchItem, workerCount int) *Dispatcher
func (*Dispatcher) Close ¶ added in v1.0.51
func (d *Dispatcher) Close()
Close cancels the worker context to abort in-flight handlers, then waits for all workers to exit. Use it to force termination after ShutdownContext times out.
func (*Dispatcher) ShutdownContext ¶ added in v1.0.51
func (d *Dispatcher) ShutdownContext(ctx context.Context) error
ShutdownContext waits for all workers to drain the event channel and exit, bounded by ctx. Workers exit once eventCh is closed and drained, so the producer must close the channel first. The worker context stays live for the whole wait so in-flight handlers can finish. On a clean drain it cancels the context group to release it. If ctx's deadline is exceeded first it returns the error without cancelling; call Close to force the workers to stop.
func (*Dispatcher) ShutdownTimeout ¶ added in v1.0.51
func (d *Dispatcher) ShutdownTimeout(timeout time.Duration) error
ShutdownTimeout calls ShutdownContext with a timeout-bounded context.
type Poller ¶ added in v1.0.51
type Poller struct {
// contains filtered or unexported fields
}
Poller owns one polling goroutine per (GithubConnectionID, OrgName) pair. It collects events from the GitHub Events API and sends them to an event channel exposed via EventCh. A separate Dispatcher consumes that channel and invokes handlers.
func (*Poller) Close ¶ added in v1.0.51
Close stops all polling goroutines, closes the event channel so the consumer can finish draining it, and shuts down the checkpoint store (a final flush). It blocks until the polling goroutines have exited. Safe to call multiple times.
func (*Poller) EventCh ¶ added in v1.0.51
func (p *Poller) EventCh() <-chan DispatchItem
EventCh returns the channel that polling goroutines send events to. A Dispatcher consumes it. The channel is closed when the Poller shuts down.
func (*Poller) TargetKeys ¶ added in v1.0.51
func (p *Poller) TargetKeys() []CheckpointKey
TargetKeys returns a snapshot of the current running target keys (for testing).
func (*Poller) UpdateTargets ¶ added in v1.0.51
func (p *Poller) UpdateTargets(desired map[CheckpointKey]ConnectionInfo)
UpdateTargets starts goroutines for new targets and stops goroutines for removed targets. It satisfies the environments.EventPoller interface.
New targets start polling in the same pass that cancels the removed ones, so they don't have to wait for the old goroutines to exit. The second pass then waits for each removed target to exit and deletes its checkpoint. Both passes run under p.mu, which is what makes the checkpoint delete safe: a cancelled target can no longer write its checkpoint once done is closed, and no concurrent UpdateTargets can re-add it between the wait and the delete. The lock order is always Poller -> Checkpoints (a polling goroutine never takes p.mu), so waiting under the lock can't deadlock.
Source Files