ctrl/subscribers.go

package steward

import (
	"fmt"
	"time"

	"github.com/prometheus/client_golang/prometheus"
)

func (s *server) subscribersStart() {
	// Start a subscriber for CLICommand messages
	{
		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)
		// fmt.Printf("*** %#v\n", proc)
		go proc.spawnWorker(s)
	}

	// 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)
		// fmt.Printf("*** %#v\n", proc)
		go proc.spawnWorker(s)
	}

	// 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)
		proc.procFuncCh = make(chan Message)
		proc.procFunc = func() error {
			sayHelloNodes := make(map[node]struct{})
			for {
				//fmt.Printf("-- DEBUG 4.1: procFunc %v, procFuncCh %v\n\n", proc.procFunc, proc.procFuncCh)
				m := <-proc.procFuncCh
				fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)
				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)),
				}
			}
		}
		go proc.spawnWorker(s)
	}

	if s.centralErrorLogger {
		// Start a subscriber for ErrorLog messages
		{
			fmt.Printf("Starting ErrorLog subscriber: %#v\n", s.nodeName)
			sub := newSubject(ErrorLog, EventNACK, "errorCentral")
			proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindSubscriber, []node{"*"}, nil)
			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)
		// TODO: Replace "central" name with variable below.
		sub := newSubject(SayHello, EventNACK, "central")
		proc := newProcess(s.processes, s.newMessagesCh, s.configuration, sub, s.errorKernel.errorCh, processKindPublisher, []node{}, nil)

		proc.procFunc = func() error {
			for {
				fmt.Printf("--- DEBUG : procFunc call:kind=%v, Subject=%v, toNode=%v\n", proc.processKind, proc.subject, proc.subject.ToNode)

				m := fmt.Sprintf("Hello from %v\n", s.nodeName)

				sam := subjectAndMessage{
					Subject: Subject{
						ToNode:         "central",
						CommandOrEvent: EventNACK,
						Method:         SayHello,
					},
					Message: Message{
						ToNode:   "central",
						FromNode: node(s.nodeName),
						Data:     []string{m},
						Method:   SayHello,
					},
				}
				proc.newMessagesCh <- []subjectAndMessage{sam}
				time.Sleep(time.Second * time.Duration(10))
			}
		}
		go proc.spawnWorker(s)
	}

	//func() {
	//	for {
	//		sam := s.createMsg(fromNode)
	//		newMessagesCh <- []subjectAndMessage{sam}
	//		time.Sleep(time.Second * time.Duration(s.interval))
	//	}
	//}()
}