mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2025-01-07 04:49:17 +00:00
Code Activity

split out start pub/sub process in separate func's

Browse source
This commit is contained in:
postmannen 2023-01-11 07:03:01 +01:00
parent d0d3ef8b39
commit 215a4c387a
1 changed files with 63 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
process.go
View file

@ -187,7 +187,25 @@ func (p process) spawnWorker() {
// Start a publisher worker, which will start a go routine (process) // Start a publisher worker, which will start a go routine (process)
// That will take care of all the messages for the subject it owns. // That will take care of all the messages for the subject it owns.
if p.processKind == processKindPublisher { if p.processKind == processKindPublisher {
p.startPublisher()
}
// Start a subscriber worker, which will start a go routine (process)
// That will take care of all the messages for the subject it owns.
if p.processKind == processKindSubscriber {
p.startSubscriber()
}
// Add information about the new process to the started processes map.
p.processes.active.mu.Lock()
p.processes.active.procNames[p.processName] = p
p.processes.active.mu.Unlock()
er := fmt.Errorf("successfully started process: %v", p.processName)
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
}
func (p process) startPublisher() {
// If there is a procFunc for the process, start it. // If there is a procFunc for the process, start it.
if p.procFunc != nil { if p.procFunc != nil {
// Initialize the channel for communication between the proc and // Initialize the channel for communication between the proc and
@ -208,9 +226,7 @@ func (p process) spawnWorker() {
go p.publishMessages(p.natsConn) go p.publishMessages(p.natsConn)
} }
// Start a subscriber worker, which will start a go routine (process) func (p process) startSubscriber() {
// That will take care of all the messages for the subject it owns.
if p.processKind == processKindSubscriber {
// If there is a procFunc for the process, start it. // If there is a procFunc for the process, start it.
if p.procFunc != nil { if p.procFunc != nil {
// Initialize the channel for communication between the proc and // Initialize the channel for communication between the proc and
@ -250,15 +266,6 @@ func (p process) spawnWorker() {
}() }()
} }
// Add information about the new process to the started processes map.
p.processes.active.mu.Lock()
p.processes.active.procNames[p.processName] = p
p.processes.active.mu.Unlock()
er := fmt.Errorf("successfully started process: %v", p.processName)
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
}
var ( var (
ErrACKSubscribeRetry = errors.New("steward: retrying to subscribe for ack message") ErrACKSubscribeRetry = errors.New("steward: retrying to subscribe for ack message")
) )