diff --git a/process.go b/process.go index 8fbd21e..8528f28 100644 --- a/process.go +++ b/process.go @@ -45,12 +45,33 @@ type process struct { processKind processKind // methodsAvailable methodsAvailable MethodsAvailable - // Helper or service function that can do some kind of work - // for the process. - // The idea is that this can hold for example the map of the - // the hello nodes to limit shared resources in the system as - // a whole for sharing a map from the *server level. - procFunc procFunc + // procFunc is a function that will be started when a worker process + // is started. If a procFunc is registered when creating a new process + // the procFunc will be started as a go routine when the process is started, + // and stopped when the process is stopped. + // + // A procFunc can be started both for publishing and subscriber processes. + // + // When used with a subscriber process the usecase is most likely to handle + // some kind of state needed for a request type. The handlers themselves + // can not hold state since they are only called once per message received, + // and exits when the message is handled leaving no state behind. With a procfunc + // we can have a process function running at all times tied to the process, and + // this function can be able to hold the state needed in a certain scenario. + // + // With a subscriber handler you generally take the message in the handler and + // pass it on to the procFunc by putting it on the procFuncCh<-, and the + // message can then be read from the procFuncCh inside the procFunc, and we + // can do some further work on it, for example update registry for metrics that + // is needed for that specific request type. + // + // With a publisher process you can attach a static function that will do some + // work to a request type, and publish the result. + // + // procFunc's can also be used to wrap in other types which we want to + // work with. An example can be handling of metrics which the message + // have no notion of, but a procFunc can have that wrapped in from when it was constructed. + procFunc func(ctx context.Context, procFuncCh chan Message) error // The channel to send a messages to the procFunc go routine. // This is typically used within the methodHandler for so we // can pass messages between the procFunc and the handler. @@ -110,34 +131,6 @@ func newProcess(ctx context.Context, metrics *metrics, natsConn *nats.Conn, proc return proc } -// procFunc is a function that will be started when a worker process -// is started. If a procFunc is registered when creating a new process -// the procFunc will be started as a go routine when the process is started, -// and stopped when the process is stopped. -// -// A procFunc can be started both for publishing and subscriber processes. -// -// When used with a subscriber process the usecase is most likely to handle -// some kind of state needed for a request type. The handlers themselves -// can not hold state since they are only called once per message received, -// and exits when the message is handled leaving no state behind. With a procfunc -// we can have a process function running at all times tied to the process, and -// this function can be able to hold the state needed in a certain scenario. -// -// With a subscriber handler you generally take the message in the handler and -// pass it on to the procFunc by putting it on the procFuncCh<-, and the -// message can then be read from the procFuncCh inside the procFunc, and we -// can do some further work on it, for example update registry for metrics that -// is needed for that specific request type. -// -// With a publisher process you can attach a static function that will do some -// work to a request type, and publish the result. -// -// procFunc's can also be used to wrap in other types which we want to -// work with. An example can be handling of metrics which the message -// have no notion of, but a procFunc can have that wrapped in from when it was constructed. -type procFunc func(ctx context.Context, procFuncCh chan Message) error - // The purpose of this function is to check if we should start a // publisher or subscriber process, where a process is a go routine // that will handle either sending or receiving messages on one diff --git a/processes.go b/processes.go index 9bf64e0..5dfd24a 100644 --- a/processes.go +++ b/processes.go @@ -254,42 +254,41 @@ func (s startup) pubREQHello(p process) { proc := newProcess(p.ctx, s.metrics, p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, processKindPublisher, nil, s.Signatures) // Define the procFunc to be used for the process. - proc.procFunc = procFunc( - func(ctx context.Context, procFuncCh chan Message) error { - ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello)) - for { + proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error { + ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello)) + for { - d := fmt.Sprintf("Hello from %v\n", p.node) + d := fmt.Sprintf("Hello from %v\n", p.node) - m := Message{ - FileName: "hello.log", - Directory: "hello-messages", - ToNode: Node(p.configuration.CentralNodeName), - FromNode: Node(p.node), - Data: []byte(d), - Method: REQHello, - ACKTimeout: 10, - Retries: 1, - } - - sam, err := newSubjectAndMessage(m) - if err != nil { - // In theory the system should drop the message before it reaches here. - p.processes.errorKernel.errSend(p, m, err) - log.Printf("error: ProcessesStart: %v\n", err) - } - proc.toRingbufferCh <- []subjectAndMessage{sam} - - select { - case <-ticker.C: - case <-ctx.Done(): - er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name()) - // sendErrorLogMessage(proc.toRingbufferCh, proc.node, er) - log.Printf("%v\n", er) - return nil - } + m := Message{ + FileName: "hello.log", + Directory: "hello-messages", + ToNode: Node(p.configuration.CentralNodeName), + FromNode: Node(p.node), + Data: []byte(d), + Method: REQHello, + ACKTimeout: 10, + Retries: 1, } - }) + + sam, err := newSubjectAndMessage(m) + if err != nil { + // In theory the system should drop the message before it reaches here. + p.processes.errorKernel.errSend(p, m, err) + log.Printf("error: ProcessesStart: %v\n", err) + } + proc.toRingbufferCh <- []subjectAndMessage{sam} + + select { + case <-ticker.C: + case <-ctx.Done(): + er := fmt.Errorf("info: stopped handleFunc for: publisher %v", proc.subject.name()) + // sendErrorLogMessage(proc.toRingbufferCh, proc.node, er) + log.Printf("%v\n", er) + return nil + } + } + } go proc.spawnWorker(p.processes, p.natsConn) } @@ -351,32 +350,31 @@ func (s startup) subREQHello(p process) { // a handler are not able to hold state, and we need to hold the state // of the nodes we've received hello's from in the sayHelloNodes map, // which is the information we pass along to generate metrics. - proc.procFunc = procFunc( - func(ctx context.Context, procFuncCh chan Message) error { - sayHelloNodes := make(map[Node]struct{}) + proc.procFunc = func(ctx context.Context, procFuncCh chan Message) error { + sayHelloNodes := make(map[Node]struct{}) - for { - // Receive a copy of the message sent from the method handler. - var m Message - - select { - case m = <-procFuncCh: - case <-ctx.Done(): - er := fmt.Errorf("info: stopped handleFunc for: subscriber %v", proc.subject.name()) - // sendErrorLogMessage(proc.toRingbufferCh, proc.node, er) - log.Printf("%v\n", er) - return nil - } - - // Add an entry for the node in the map - sayHelloNodes[m.FromNode] = struct{}{} - - // update the prometheus metrics - s.metrics.promHelloNodesTotal.Set(float64(len(sayHelloNodes))) - s.metrics.promHelloNodesContactLast.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime() + for { + // Receive a copy of the message sent from the method handler. + var m Message + select { + case m = <-procFuncCh: + case <-ctx.Done(): + er := fmt.Errorf("info: stopped handleFunc for: subscriber %v", proc.subject.name()) + // sendErrorLogMessage(proc.toRingbufferCh, proc.node, er) + log.Printf("%v\n", er) + return nil } - }) + + // Add an entry for the node in the map + sayHelloNodes[m.FromNode] = struct{}{} + + // update the prometheus metrics + s.metrics.promHelloNodesTotal.Set(float64(len(sayHelloNodes))) + s.metrics.promHelloNodesContactLast.With(prometheus.Labels{"nodeName": string(m.FromNode)}).SetToCurrentTime() + + } + } go proc.spawnWorker(p.processes, p.natsConn) }