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

Moved out the error handling from the process run

Browse source
This commit is contained in:
postmannen 2021-02-01 13:41:04 +01:00
parent 223f472f5b
commit ec76bd36cd
2 changed files with 27 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
publisher.go
View file

@ -56,7 +56,7 @@ type server struct {
processes map[node]process
// The last processID created
lastProcessID int
thisNodeName string
nodeName string
}
// newServer will prepare and return a server type
@ -66,23 +66,12 @@ func NewServer(brokerAddress string, nodeName string) (*server, error) {
log.Printf("error: nats.Connect failed: %v\n", err)
}
return &server{
thisNodeName: nodeName,
natsConn: conn,
processes: make(map[node]process),
}, nil
}
s := &server{
nodeName: nodeName,
natsConn: conn,
processes: make(map[node]process),
}
func (s *server) RunPublisher() {
proc := s.prepareNewProcess("btship1")
// fmt.Printf("*** %#v\n", proc)
go s.spawnProcess(proc)
proc = s.prepareNewProcess("btship2")
// fmt.Printf("*** %#v\n", proc)
go s.spawnProcess(proc)
// start the error handling
go func() {
for {
@ -98,6 +87,19 @@ func (s *server) RunPublisher() {
}
}()
return s, nil
}
func (s *server) RunPublisher() {
proc := s.prepareNewProcess("btship1")
// fmt.Printf("*** %#v\n", proc)
go s.spawnProcess(proc)
proc = s.prepareNewProcess("btship2")
// fmt.Printf("*** %#v\n", proc)
go s.spawnProcess(proc)
select {}
}
@ -118,6 +120,8 @@ type process struct {
errorCh chan string
}
// prepareNewProcess will set the the provided values and the default
// values for a process.
func (s *server) prepareNewProcess(nodeName string) process {
// create the initial configuration for a sessions communicating with 1 host.
s.lastProcessID++
@ -131,7 +135,9 @@ func (s *server) prepareNewProcess(nodeName string) process {
return proc
}
// spawnProcess will spawn a new process
// spawnProcess will spawn a new process. It will give the process
// the next available ID, and also add the process to the processes
// map.
func (s *server) spawnProcess(proc process) {
mu.Lock()
s.processes[proc.node] = proc
@ -151,7 +157,7 @@ func (s *server) spawnProcess(proc process) {
// simulate that we get an error, and that we can send that
// out of the process and receive it in another thread.
// s.processes[proc.node].errorCh <- "received an error from process: " + fmt.Sprintf("%v\n", proc.processID)
s.processes[proc.node].errorCh <- "received an error from process: " + fmt.Sprintf("%v\n", proc.processID)
//fmt.Printf("%#v\n", s.processes[proc.node])
}

4
subscriber.go
View file

@ -18,8 +18,8 @@ func (s *server) RunSubscriber() {
// Subscribe will start up a Go routine under the hood calling the
// callback function specified when a new message is received.
subject := fmt.Sprintf("%s.%s.%s", s.thisNodeName, "command", "shellcommand")
_, err := s.natsConn.Subscribe(subject, listenForMessage(s.natsConn, reqMsgCh, s.thisNodeName))
subject := fmt.Sprintf("%s.%s.%s", s.nodeName, "command", "shellcommand")
_, err := s.natsConn.Subscribe(subject, listenForMessage(s.natsConn, reqMsgCh, s.nodeName))
if err != nil {
fmt.Printf("error: Subscribe failed: %v\n", err)
}