mirrors/ctrl
mirrors/ctrl
1
0
Fork 0
mirror of https://github.com/postmannen/ctrl.git synced 2024-12-15 17:51:15 +00:00
Code Activity

moved all starting of pub/sub procs under process

Browse source
This commit is contained in:
postmannen 2021-04-08 13:43:47 +02:00
parent ab49f5a4de
commit c091ebfa38
3 changed files with 100 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
server.go
View file

@ -131,7 +131,9 @@ func (s *server) Start() {
go s.readSocket(s.toRingbufferCh) go s.readSocket(s.toRingbufferCh)
// Start up the predefined subscribers. // Start up the predefined subscribers.
s.ProcessesStart() sub := newSubject(REQInitial, s.nodeName)
p := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, "", []node{}, nil)
p.ProcessesStart()
time.Sleep(time.Second * 1) time.Sleep(time.Second * 1)
s.processes.printProcessesMap() s.processes.printProcessesMap()

136
startup_processes.go
View file

@ -9,45 +9,45 @@ import (
"github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus"
) )
func (s *server) ProcessesStart() { func (p process) ProcessesStart() {
// --- Subscriber services that can be started via flags // --- Subscriber services that can be started via flags
{ {
fmt.Printf("Starting REQOpCommand subscriber: %#v\n", s.nodeName) fmt.Printf("Starting REQOpCommand subscriber: %#v\n", p.node)
sub := newSubject(REQOpCommand, s.nodeName) sub := newSubject(REQOpCommand, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, []node{"*"}, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
// Start a subscriber for textLogging messages // Start a subscriber for textLogging messages
if s.configuration.StartSubREQTextToLogFile.OK { if p.configuration.StartSubREQTextToLogFile.OK {
{ {
fmt.Printf("Starting text logging subscriber: %#v\n", s.nodeName) fmt.Printf("Starting text logging subscriber: %#v\n", p.node)
sub := newSubject(REQTextToLogFile, s.nodeName) sub := newSubject(REQTextToLogFile, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQTextToLogFile.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToLogFile.Values, nil)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for text to file messages // Start a subscriber for text to file messages
if s.configuration.StartSubREQTextToFile.OK { if p.configuration.StartSubREQTextToFile.OK {
{ {
fmt.Printf("Starting text to file subscriber: %#v\n", s.nodeName) fmt.Printf("Starting text to file subscriber: %#v\n", p.node)
sub := newSubject(REQTextToFile, s.nodeName) sub := newSubject(REQTextToFile, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQTextToFile.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToFile.Values, nil)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for Hello messages // Start a subscriber for Hello messages
if s.configuration.StartSubREQHello.OK { if p.configuration.StartSubREQHello.OK {
{ {
fmt.Printf("Starting Hello subscriber: %#v\n", s.nodeName) fmt.Printf("Starting Hello subscriber: %#v\n", p.node)
sub := newSubject(REQHello, s.nodeName) sub := newSubject(REQHello, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQHello.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQHello.Values, nil)
proc.procFuncCh = make(chan Message) proc.procFuncCh = make(chan Message)
// The reason for running the say hello subscriber as a procFunc is that // The reason for running the say hello subscriber as a procFunc is that
@ -82,67 +82,67 @@ func (s *server) ProcessesStart() {
} }
} }
} }
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
if s.configuration.StartSubREQErrorLog.OK { if p.configuration.StartSubREQErrorLog.OK {
// Start a subscriber for REQErrorLog messages // Start a subscriber for REQErrorLog messages
{ {
fmt.Printf("Starting REQErrorLog subscriber: %#v\n", s.nodeName) fmt.Printf("Starting REQErrorLog subscriber: %#v\n", p.node)
sub := newSubject(REQErrorLog, "errorCentral") sub := newSubject(REQErrorLog, "errorCentral")
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQErrorLog.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQErrorLog.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for Ping Request messages // Start a subscriber for Ping Request messages
if s.configuration.StartSubREQPing.OK { if p.configuration.StartSubREQPing.OK {
{ {
fmt.Printf("Starting Ping Request subscriber: %#v\n", s.nodeName) fmt.Printf("Starting Ping Request subscriber: %#v\n", p.node)
sub := newSubject(REQPing, s.nodeName) sub := newSubject(REQPing, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQPing.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQPing.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for REQPong messages // Start a subscriber for REQPong messages
if s.configuration.StartSubREQPong.OK { if p.configuration.StartSubREQPong.OK {
{ {
fmt.Printf("Starting Pong subscriber: %#v\n", s.nodeName) fmt.Printf("Starting Pong subscriber: %#v\n", p.node)
sub := newSubject(REQPong, s.nodeName) sub := newSubject(REQPong, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQPong.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQPong.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for REQCliCommand messages // Start a subscriber for REQCliCommand messages
if s.configuration.StartSubREQCliCommand.OK { if p.configuration.StartSubREQCliCommand.OK {
{ {
fmt.Printf("Starting CLICommand Request subscriber: %#v\n", s.nodeName) fmt.Printf("Starting CLICommand Request subscriber: %#v\n", p.node)
sub := newSubject(REQCliCommand, s.nodeName) sub := newSubject(REQCliCommand, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQCliCommand.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQCliCommand.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for Not In Order Cli Command Request messages // Start a subscriber for Not In Order Cli Command Request messages
if s.configuration.StartSubREQnCliCommand.OK { if p.configuration.StartSubREQnCliCommand.OK {
{ {
fmt.Printf("Starting CLICommand Not Sequential Request subscriber: %#v\n", s.nodeName) fmt.Printf("Starting CLICommand Not Sequential Request subscriber: %#v\n", p.node)
sub := newSubject(REQnCliCommand, s.nodeName) sub := newSubject(REQnCliCommand, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQnCliCommand.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQnCliCommand.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
// Start a subscriber for CLICommandReply messages // Start a subscriber for CLICommandReply messages
if s.configuration.StartSubREQTextToConsole.OK { if p.configuration.StartSubREQTextToConsole.OK {
{ {
fmt.Printf("Starting Text To Console subscriber: %#v\n", s.nodeName) fmt.Printf("Starting Text To Console subscriber: %#v\n", p.node)
sub := newSubject(REQTextToConsole, s.nodeName) sub := newSubject(REQTextToConsole, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQTextToConsole.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQTextToConsole.Values, nil)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
@ -152,24 +152,24 @@ func (s *server) ProcessesStart() {
// Define a process of kind publisher with subject for SayHello to central, // Define a process of kind publisher with subject for SayHello to central,
// and register a procFunc with the process that will handle the actual // and register a procFunc with the process that will handle the actual
// sending of say hello. // sending of say hello.
if s.configuration.StartPubREQHello != 0 { if p.configuration.StartPubREQHello != 0 {
fmt.Printf("Starting Hello Publisher: %#v\n", s.nodeName) fmt.Printf("Starting Hello Publisher: %#v\n", p.node)
sub := newSubject(REQHello, s.configuration.CentralNodeName) sub := newSubject(REQHello, p.configuration.CentralNodeName)
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, []node{}, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindPublisher, []node{}, nil)
// Define the procFunc to be used for the process. // Define the procFunc to be used for the process.
proc.procFunc = procFunc( proc.procFunc = procFunc(
func(ctx context.Context) error { func(ctx context.Context) error {
ticker := time.NewTicker(time.Second * time.Duration(s.configuration.StartPubREQHello)) ticker := time.NewTicker(time.Second * time.Duration(p.configuration.StartPubREQHello))
for { for {
fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode) fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)
d := fmt.Sprintf("Hello from %v\n", s.nodeName) d := fmt.Sprintf("Hello from %v\n", p.node)
m := Message{ m := Message{
ToNode: "central", ToNode: "central",
FromNode: node(s.nodeName), FromNode: node(p.node),
Data: []string{d}, Data: []string{d},
Method: REQHello, Method: REQHello,
} }
@ -191,17 +191,29 @@ func (s *server) ProcessesStart() {
} }
} }
}) })
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
// Start a subscriber for Http Get Requests // Start a subscriber for Http Get Requests
if s.configuration.StartSubREQHttpGet.OK { if p.configuration.StartSubREQHttpGet.OK {
{ {
fmt.Printf("Starting Http Get subscriber: %#v\n", s.nodeName) fmt.Printf("Starting Http Get subscriber: %#v\n", p.node)
sub := newSubject(REQHttpGet, s.nodeName) sub := newSubject(REQHttpGet, string(p.node))
proc := newProcess(s.natsConn, s.processes, s.toRingbufferCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, s.configuration.StartSubREQHttpGet.Values, nil) proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQHttpGet.Values, nil)
// fmt.Printf("*** %#v\n", proc) // fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(s.processes, s.natsConn) go proc.spawnWorker(p.processes, p.natsConn)
} }
} }
} }
type startups struct{}
func (s startups) REQHttpGet(p process) {
{
fmt.Printf("Starting Http Get subscriber: %#v\n", p.node)
sub := newSubject(REQHttpGet, string(p.node))
proc := newProcess(p.natsConn, p.processes, p.toRingbufferCh, p.configuration, sub, p.errorCh, processKindSubscriber, p.configuration.StartSubREQHttpGet.Values, nil)
// fmt.Printf("*** %#v\n", proc)
go proc.spawnWorker(p.processes, p.natsConn)
}
}

23
subscriber_method_types.go
View file

@ -52,6 +52,8 @@ type Method string
// The constants that will be used throughout the system for // The constants that will be used throughout the system for
// when specifying what kind of Method to send or work with. // when specifying what kind of Method to send or work with.
const ( const (
// Initial method used to start other processes.
REQInitial Method = "REQInitial"
// Command for client operation request of the system. The op // Command for client operation request of the system. The op
// command to execute shall be given in the data field of the // command to execute shall be given in the data field of the
// message as string value. For example "ps". // message as string value. For example "ps".
@ -123,6 +125,9 @@ func (m Method) GetMethodsAvailable() MethodsAvailable {
// Event, Used to communicate that an action has been performed. // Event, Used to communicate that an action has been performed.
ma := MethodsAvailable{ ma := MethodsAvailable{
methodhandlers: map[Method]methodHandler{ methodhandlers: map[Method]methodHandler{
REQInitial: methodREQInitial{
commandOrEvent: CommandACK,
},
REQOpCommand: methodREQOpCommand{ REQOpCommand: methodREQOpCommand{
commandOrEvent: CommandACK, commandOrEvent: CommandACK,
}, },
@ -175,6 +180,24 @@ func (m Method) getHandler(method Method) methodHandler {
// The structure that works as a reference for all the methods and if // The structure that works as a reference for all the methods and if
// they are of the command or event type, and also if it is a ACK or // they are of the command or event type, and also if it is a ACK or
// NACK message. // NACK message.
// ----
type methodREQInitial struct {
commandOrEvent CommandOrEvent
}
func (m methodREQInitial) getKind() CommandOrEvent {
return m.commandOrEvent
}
func (m methodREQInitial) handler(proc process, message Message, node string) ([]byte, error) {
// proc.procFuncCh <- message
ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID))
return ackMsg, nil
}
// ----
type MethodsAvailable struct { type MethodsAvailable struct {
methodhandlers map[Method]methodHandler methodhandlers map[Method]methodHandler
} }