-
Notifications
You must be signed in to change notification settings - Fork 29
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #522 from ripienaar/consumer_pause
Adds the ability to pause and resume consumers
- Loading branch information
Showing
22 changed files
with
498 additions
and
22 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package advisory | ||
|
|
||
| import ( | ||
| "time" | ||
|
|
||
| "github.com/nats-io/jsm.go/api/event" | ||
| ) | ||
|
|
||
| // JSConsumerPauseAdvisoryV1 indicates that a consumer was paused or unpaused | ||
| type JSConsumerPauseAdvisoryV1 struct { | ||
| event.NATSEvent | ||
|
|
||
| Stream string `json:"stream"` | ||
| Consumer string `json:"consumer"` | ||
| Paused bool `json:"paused"` | ||
| PauseUntil time.Time `json:"pause_until,omitempty"` | ||
| Domain string `json:"domain,omitempty"` | ||
| } | ||
|
|
||
| func init() { | ||
| err := event.RegisterTextCompactTemplate("io.nats.jetstream.advisory.v1.consumer_pause", `{{ .Time | ShortTime }} [Consumer Pause] Consumer: {{ .Stream }} > {{ .Consumer }} Paused: {{ .Paused }}{{ if .Paused }} until {{ .PauseUntil }}{{ end }}`) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| err = event.RegisterTextExtendedTemplate("io.nats.jetstream.advisory.v1.consumer_pause", ` | ||
| [{{ .Time | ShortTime }}] [{{ .ID }}] Consumer Pause | ||
| Stream: {{ .Stream }} | ||
| Consumer: {{ .Consumer }} | ||
| Paused: {{ .Paused }} | ||
| {{- if .Paused }} | ||
| Until: {{ .PauseUntil }} | ||
| {{- end }} | ||
| {{- if .Domain }} | ||
| Domain: {{ .Domain }} | ||
| {{- end }} | ||
| `) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| { | ||
| "$schema": "http://json-schema.org/draft-07/schema#", | ||
| "$id": "https://nats.io/schemas/jetstream/advisory/v1/consumer_pause.json", | ||
| "description": "An Advisory sent when consumer is paused or resumed", | ||
| "title": "io.nats.jetstream.advisory.v1.consumer_pause", | ||
| "type": "object", | ||
| "required": [ | ||
| "type", | ||
| "id", | ||
| "timestamp", | ||
| "stream", | ||
| "consumer", | ||
| "paused" | ||
| ], | ||
| "additionalProperties": false, | ||
| "properties": { | ||
| "type": { | ||
| "type": "string", | ||
| "const": "io.nats.jetstream.advisory.v1.consumer_pause" | ||
| }, | ||
| "id": { | ||
| "type": "string", | ||
| "description": "Unique correlation ID for this event" | ||
| }, | ||
| "timestamp": { | ||
| "type": "string", | ||
| "description": "The time this event was created in RFC3339 format" | ||
| }, | ||
| "stream": { | ||
| "type": "string", | ||
| "description": "The name of the Stream the Consumer belongs to" | ||
| }, | ||
| "consumer": { | ||
| "type": "string", | ||
| "description": "The name of the Consumer that elected a new leader" | ||
| }, | ||
| "paused": { | ||
| "type": "boolean", | ||
| "description": "Indicates the consumer is paused" | ||
| }, | ||
| "pause_until": { | ||
| "description": "When paused the time the consumer will be unpaused, RFC3339 format", | ||
| "type": "string" | ||
| }, | ||
| "domain": { | ||
| "description": "The domain hosting the Stream and Consumer if configured", | ||
| "type": "string" | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.