mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2025-03-15 10:57:42 +00:00
Code Activity

comments

Browse source
This commit is contained in:
postmannen 2021-08-12 09:21:56 +02:00
parent a0304e31e7
commit 3fb9aefd78
2 changed files with 7 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
processes.go
View file

@ -27,6 +27,8 @@ type processes struct {
promTotalProcesses prometheus.Gauge promTotalProcesses prometheus.Gauge
// //
promProcessesVec *prometheus.GaugeVec promProcessesVec *prometheus.GaugeVec
// Waitgroup to keep track of all the processes started
wg sync.WaitGroup
} }
// newProcesses will prepare and return a *processes which // newProcesses will prepare and return a *processes which
@ -69,7 +71,7 @@ func (p *processes) Start(proc process) {
{ {
log.Printf("Starting REQOpCommand subscriber: %#v\n", proc.node) log.Printf("Starting REQOpCommand subscriber: %#v\n", proc.node)
sub := newSubject(REQOpCommand, string(proc.node)) sub := newSubject(REQOpCommand, string(proc.node))
proc := newProcess(proc.ctx, proc.natsConn, proc.processes, proc.toRingbufferCh, proc.configuration, sub, proc.errorCh, processKindSubscriber, []Node{Node(proc.configuration.CentralNodeName)}, nil) proc := newProcess(proc.ctx, proc.natsConn, p, proc.toRingbufferCh, proc.configuration, sub, proc.errorCh, processKindSubscriber, []Node{Node(proc.configuration.CentralNodeName)}, nil)
go proc.spawnWorker(proc.processes, proc.natsConn) go proc.spawnWorker(proc.processes, proc.natsConn)
} }

20
subscriber_method_types.go
View file

@ -358,20 +358,12 @@ func (m methodREQOpCommand) handler(proc process, message Message, nodeName stri
// Assert it into the correct non pointer value. // Assert it into the correct non pointer value.
arg := *dst.(*OpCmdStopProc) arg := *dst.(*OpCmdStopProc)
// Data layout: OPCommand, Method, publisher/subscriber, receivingNode // Based on the arg values received in the message we create can
// // create a processName structure as used in naming the real processes.
// The processes can be either publishers or subscribers. The subject name // We can then use this processName to get the real values for the
// are used within naming a process. Since the subject structure contains // actual process we want to stop.
// the node name of the node that will receive this message we also need
// specify it so we are able to delete the publisher processes, since a
// publisher process will have the name of the node to receive the message,
// and not just the local node name as with subscriber processes.
// receive the message we need to specify
// Process name example: ship2.REQToFileAppend.EventACK_subscriber
sub := newSubject(arg.Method, string(arg.RecevingNode)) sub := newSubject(arg.Method, string(arg.RecevingNode))
processName := processNameGet(sub.name(), arg.Kind) processName := processNameGet(sub.name(), arg.Kind)
// fmt.Printf(" ** DEBUG1: processName: %v\n", processName)
// Check if the message contains an id. // Check if the message contains an id.
err = func() error { err = func() error {
@ -392,10 +384,6 @@ func (m methodREQOpCommand) handler(proc process, message Message, nodeName stri
proc.processes.mu.Lock() proc.processes.mu.Lock()
// for k, v := range proc.processes.active {
// fmt.Printf(" ** DEBUG1.3: MAP: k = %v, v = %v\n", k, v.processKind)
// }
toStopProc, ok := proc.processes.active[processName][arg.ID] toStopProc, ok := proc.processes.active[processName][arg.ID]
if ok { if ok {
// Delete the process from the processes map // Delete the process from the processes map