1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-14 12:37:31 +00:00
ctrl/runProcessesAtStartup.go

146 lines
5.3 KiB
Go
Raw Normal View History

2021-02-24 09:58:02 +00:00
package steward
import (
"fmt"
"log"
"time"
"github.com/prometheus/client_golang/prometheus"
2021-02-24 09:58:02 +00:00
)
2021-03-10 06:11:14 +00:00
func (s *server) ProcessesStart() {
// Start a subscriber for CLICommand messages
2021-02-24 09:58:02 +00:00
{
fmt.Printf("Starting CLICommand subscriber: %#v\n", s.nodeName)
sub := newSubject(CLICommand, CommandACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"central", "ship2"}, nil)
2021-02-24 09:58:02 +00:00
// fmt.Printf("*** %#v\n", proc)
2021-03-03 13:14:32 +00:00
go proc.spawnWorker(s)
2021-02-24 09:58:02 +00:00
}
// Start a subscriber for textLogging messages
{
fmt.Printf("Starting textlogging subscriber: %#v\n", s.nodeName)
sub := newSubject(TextLogging, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
2021-02-24 09:58:02 +00:00
// fmt.Printf("*** %#v\n", proc)
2021-03-03 13:14:32 +00:00
go proc.spawnWorker(s)
2021-02-24 09:58:02 +00:00
}
// Start a subscriber for SayHello messages
{
fmt.Printf("Starting SayHello subscriber: %#v\n", s.nodeName)
sub := newSubject(SayHello, EventNACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
2021-03-05 12:10:46 +00:00
proc.procFuncCh = make(chan Message)
2021-03-10 06:11:14 +00:00
2021-03-05 12:10:46 +00:00
proc.procFunc = func() error {
sayHelloNodes := make(map[node]struct{})
for {
2021-03-10 13:14:09 +00:00
// Receive a copy of the message sent from the method handler.
2021-03-05 12:10:46 +00:00
m := <-proc.procFuncCh
fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)
2021-03-10 13:14:09 +00:00
2021-03-05 12:10:46 +00:00
sayHelloNodes[m.FromNode] = struct{}{}
// update the prometheus metrics
s.metrics.metricsCh <- metricType{
metric: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "hello_nodes",
Help: "The current number of total nodes who have said hello",
}),
value: float64(len(sayHelloNodes)),
}
}
}
2021-03-03 13:14:32 +00:00
go proc.spawnWorker(s)
2021-02-24 09:58:02 +00:00
}
2021-03-25 11:50:58 +00:00
if s.configuration.StartSubErrorLog.OK {
// Start a subscriber for ErrorLog messages
{
fmt.Printf("Starting ErrorLog subscriber: %#v\n", s.nodeName)
sub := newSubject(ErrorLog, EventNACK, "errorCentral")
2021-03-25 11:50:58 +00:00
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubErrorLog.Values, nil)
2021-03-03 13:14:32 +00:00
go proc.spawnWorker(s)
}
}
// --------- Testing with publisher ------------
// Define a process of kind publisher with subject for SayHello to central,
// and register a procFunc with the process that will handle the actual
// sending of say hello.
if s.configuration.PublisherServiceSayhello != 0 {
fmt.Printf("Starting SayHello Publisher: %#v\n", s.nodeName)
2021-03-10 13:14:09 +00:00
sub := newSubject(SayHello, EventNACK, s.configuration.CentralNodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, []node{}, nil)
2021-03-10 13:14:09 +00:00
// Define the procFun to be used for the process.
proc.procFunc = procFunc(
func() error {
for {
fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)
2021-03-10 13:14:09 +00:00
d := fmt.Sprintf("Hello from %v\n", s.nodeName)
2021-03-10 13:14:09 +00:00
m := Message{
ToNode: "central",
FromNode: node(s.nodeName),
Data: []string{d},
Method: SayHello,
}
2021-03-10 06:11:14 +00:00
sam, err := newSAM(m)
if err != nil {
// In theory the system should drop the message before it reaches here.
log.Printf("error: ProcessesStart: %v\n", err)
}
2021-03-10 13:14:09 +00:00
proc.newMessagesCh <- []subjectAndMessage{sam}
time.Sleep(time.Second * time.Duration(s.configuration.PublisherServiceSayhello))
}
})
go proc.spawnWorker(s)
}
2021-03-11 05:34:36 +00:00
// Start a subscriber for ECHORequest messages
{
fmt.Printf("Starting Echo Request subscriber: %#v\n", s.nodeName)
sub := newSubject(ECHORequest, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s)
}
// Start a subscriber for ECHOReply messages
{
fmt.Printf("Starting Echo Reply subscriber: %#v\n", s.nodeName)
sub := newSubject(ECHOReply, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s)
}
// Start a subscriber for CLICommandRequest messages
{
fmt.Printf("Starting CLICommand Request subscriber: %#v\n", s.nodeName)
sub := newSubject(CLICommandRequest, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s)
}
// Start a subscriber for CLICommandRequest messages
{
fmt.Printf("Starting CLICommand NOSEQ Request subscriber: %#v\n", s.nodeName)
sub := newSubject(CLICommandRequestNOSEQ, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s)
}
// Start a subscriber for CLICommandReply messages
{
fmt.Printf("Starting CLICommand Reply subscriber: %#v\n", s.nodeName)
sub := newSubject(CLICommandReply, EventACK, s.nodeName)
proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s)
}
2021-02-24 09:58:02 +00:00
}