-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Major changes to how runners work, implicit service runners.
- Loading branch information
Showing
14 changed files
with
481 additions
and
348 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
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 |
|---|---|---|
| @@ -1,99 +1,39 @@ | ||
| package enats | ||
|
|
||
| import ( | ||
| "strconv" | ||
| "strings" | ||
|
|
||
| "get.pme.sh/pmesh/config" | ||
| "get.pme.sh/pmesh/xlog" | ||
| ) | ||
|
|
||
| var AnyCastMachineID = config.MachineID(0) | ||
| var remoteQueueName = "ev." + AnyCastMachineID.String() + "." | ||
| var localQueueName = "ev." + config.GetMachineID().String() + "." | ||
| const ( | ||
| userPrefix = "raw." | ||
| ) | ||
|
|
||
| // Given a nats subject, return the user-specified topic. | ||
| func ToTopic(subject string) (topic string, dest config.MachineID) { | ||
| dest = AnyCastMachineID | ||
| func ToTopic(subject string) string { | ||
| return strings.TrimPrefix(subject, EventStreamPrefix) | ||
| } | ||
|
|
||
| // Given a user-specified topic, return the subject names to use for NATS consumers/publishers. | ||
| func ToSubject(topic string) string { | ||
| // User specified topic | ||
| if strings.HasPrefix(subject, "jet.") { | ||
| topic = strings.TrimPrefix(subject, "jet.") | ||
| return | ||
| } | ||
|
|
||
| // Remote topic | ||
| if strings.HasPrefix(subject, remoteQueueName) { | ||
| topic = subject[len(remoteQueueName):] | ||
| return | ||
| } | ||
|
|
||
| // Local topic | ||
| if strings.HasPrefix(subject, localQueueName) { | ||
| topic = subject[len(localQueueName):] | ||
| dest = config.GetMachineID() | ||
| return | ||
| } | ||
|
|
||
| if strings.HasPrefix(subject, "ev.") { | ||
| if len(subject) <= len(remoteQueueName) || subject[len(remoteQueueName)-1] != '.' { | ||
| xlog.Warn().Str("subject", subject).Msg("Invalid subject") | ||
| return | ||
| } | ||
|
|
||
| u64, err := strconv.ParseUint(subject[len("ev."):len(remoteQueueName)-1], 16, 32) | ||
| if err != nil { | ||
| xlog.Warn().Str("subject", subject).Err(err).Msg("Invalid subject") | ||
| return | ||
| } | ||
|
|
||
| dest = config.MachineID(uint32(u64)) | ||
| topic = subject[len(remoteQueueName):] | ||
| return | ||
| if subject, ok := strings.CutPrefix(topic, userPrefix); ok { | ||
| topic = subject | ||
| } else if !strings.HasPrefix(topic, EventStreamPrefix) { | ||
| topic = EventStreamPrefix + topic | ||
| } | ||
|
|
||
| topic = subject | ||
| return | ||
| } | ||
|
|
||
| // Given a user-specified topic, return the subject names to use for NATS consumers. | ||
| func ToConsumerSubjects(topic string) []string { | ||
| // Wildcard | ||
| if strings.HasSuffix(topic, ".") { | ||
| topic += ">" | ||
| } | ||
| if strings.HasPrefix(topic, "jet.") { | ||
| return []string{strings.TrimPrefix(topic, "jet.")} | ||
| } | ||
| if strings.HasPrefix(topic, "$local.") { | ||
| return []string{localQueueName + strings.TrimPrefix(topic, "$local.")} | ||
| } | ||
| return []string{ | ||
| remoteQueueName + topic, | ||
| localQueueName + topic, | ||
| } | ||
| } | ||
|
|
||
| // Given a user-specified topic, return the subject name to use for NATS publishers. | ||
| func ToPublisherSubject(topic string) string { | ||
| if strings.HasPrefix(topic, "jet.") { | ||
| return strings.TrimPrefix(topic, "jet.") | ||
| } else if !strings.HasPrefix(topic, "$local.") { | ||
| return remoteQueueName + topic | ||
| } else { | ||
| return localQueueName + strings.TrimPrefix(topic, "$local.") | ||
| } | ||
| } | ||
| func ToPublisherSubjectWithTarget(topic string, target config.MachineID) string { | ||
| if strings.HasPrefix(topic, "jet.") { | ||
| return strings.TrimPrefix(topic, "jet.") | ||
| } else { | ||
| return "ev." + config.GetMachineID().String() + "." + strings.TrimPrefix(topic, "$local.") | ||
| } | ||
| return topic | ||
| } | ||
|
|
||
| // Given a user-specified topic, return the queue name to use for NATS consumers. | ||
| func ToConsumerQueueName(pfx, topic string) string { | ||
| queue := strings.ReplaceAll(pfx+topic, ".", "-") | ||
| queue = strings.ReplaceAll(queue, "*", "all") | ||
| queue = strings.ReplaceAll(queue, ">", "matchall") | ||
| queue := pfx + ToSubject(topic) | ||
| queue = strings.ReplaceAll(queue, ".", "-") | ||
| queue = strings.ReplaceAll(queue, "*", "any") | ||
| queue = strings.ReplaceAll(queue, ">", "all") | ||
| return queue | ||
| } |
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
Oops, something went wrong.