mirror of
https://github.com/postmannen/ctrl.git
synced 2024-12-14 12:37:31 +00:00
replaced log lines with slog from errorKernel
This commit is contained in:
postmannen
parent
1dbff78c81
commit
7fcf4b1077
3 changed files with 125 additions and 69 deletions
|
|
@ -75,7 +75,8 @@ func newPKI(configuration *Configuration, errorKernel *errorKernel) *pki {
|
|||
// Open the database file for persistent storage of public keys.
|
||||
db, err := bolt.Open(databaseFilepath, 0660, nil)
|
||||
if err != nil {
|
||||
log.Printf("error: failed to open db: %v\n", err)
|
||||
er := fmt.Errorf("newPKI: error: failed to open db: %v", err)
|
||||
errorKernel.logConsoleOnlyIfDebug(er, configuration)
|
||||
return &p
|
||||
}
|
||||
|
||||
|
|
@ -84,14 +85,16 @@ func newPKI(configuration *Configuration, errorKernel *errorKernel) *pki {
|
|||
// Get public keys from db storage.
|
||||
keys, err := p.dbDumpPublicKey()
|
||||
if err != nil {
|
||||
log.Printf("debug: dbPublicKeyDump failed, probably empty db: %v\n", err)
|
||||
er := fmt.Errorf("newPKI: dbPublicKeyDump failed, probably empty db: %v", err)
|
||||
errorKernel.logConsoleOnlyIfDebug(er, configuration)
|
||||
}
|
||||
|
||||
// Only assign from storage to in memory map if the storage contained any values.
|
||||
if keys != nil {
|
||||
p.nodesAcked.keysAndHash.Keys = keys
|
||||
for k, v := range keys {
|
||||
log.Printf("info: public keys db contains: %v, %v\n", k, []byte(v))
|
||||
er := fmt.Errorf("newPKI: public keys db contains: %v, %v", k, []byte(v))
|
||||
errorKernel.logConsoleOnlyIfDebug(er, configuration)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
45
node_auth.go
45
node_auth.go
|
|
@ -42,8 +42,8 @@ type nodeAuth struct {
|
|||
|
||||
func newNodeAuth(configuration *Configuration, errorKernel *errorKernel) *nodeAuth {
|
||||
n := nodeAuth{
|
||||
nodeAcl: newNodeAcl(configuration),
|
||||
publicKeys: newPublicKeys(configuration),
|
||||
nodeAcl: newNodeAcl(configuration, errorKernel),
|
||||
publicKeys: newPublicKeys(configuration, errorKernel),
|
||||
configuration: configuration,
|
||||
errorKernel: errorKernel,
|
||||
}
|
||||
|
|
@ -79,15 +79,19 @@ func newAclAndHash() aclAndHash {
|
|||
|
||||
type nodeAcl struct {
|
||||
// allowed is a map for holding all the allowed signatures.
|
||||
aclAndHash aclAndHash
|
||||
filePath string
|
||||
mu sync.Mutex
|
||||
aclAndHash aclAndHash
|
||||
filePath string
|
||||
mu sync.Mutex
|
||||
errorKernel *errorKernel
|
||||
configuration *Configuration
|
||||
}
|
||||
|
||||
func newNodeAcl(c *Configuration) *nodeAcl {
|
||||
func newNodeAcl(c *Configuration, errorKernel *errorKernel) *nodeAcl {
|
||||
n := nodeAcl{
|
||||
aclAndHash: newAclAndHash(),
|
||||
filePath: filepath.Join(c.DatabaseFolder, "node_aclmap.txt"),
|
||||
aclAndHash: newAclAndHash(),
|
||||
filePath: filepath.Join(c.DatabaseFolder, "node_aclmap.txt"),
|
||||
errorKernel: errorKernel,
|
||||
configuration: c,
|
||||
}
|
||||
|
||||
err := n.loadFromFile()
|
||||
|
|
@ -106,7 +110,8 @@ func (n *nodeAcl) loadFromFile() error {
|
|||
if _, err := os.Stat(n.filePath); os.IsNotExist(err) {
|
||||
// Just logging the error since it is not crucial that a key file is missing,
|
||||
// since a new one will be created on the next update.
|
||||
log.Printf("no acl file found at %v\n", n.filePath)
|
||||
er := fmt.Errorf("acl: loadFromFile: no acl file found at %v", n.filePath)
|
||||
n.errorKernel.logConsoleOnlyIfDebug(er, n.configuration)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -128,7 +133,8 @@ func (n *nodeAcl) loadFromFile() error {
|
|||
return err
|
||||
}
|
||||
|
||||
log.Printf("\n ***** DEBUG: Loaded existing acl's from file: %v\n\n", n.aclAndHash.Hash)
|
||||
er := fmt.Errorf("nodeAcl: loadFromFile: Loaded existing acl's from file: %v", n.aclAndHash.Hash)
|
||||
n.errorKernel.logConsoleOnlyIfDebug(er, n.configuration)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
@ -178,15 +184,19 @@ func newKeysAndHash() *keysAndHash {
|
|||
}
|
||||
|
||||
type publicKeys struct {
|
||||
keysAndHash *keysAndHash
|
||||
mu sync.Mutex
|
||||
filePath string
|
||||
keysAndHash *keysAndHash
|
||||
mu sync.Mutex
|
||||
filePath string
|
||||
errorKernel *errorKernel
|
||||
configuration *Configuration
|
||||
}
|
||||
|
||||
func newPublicKeys(c *Configuration) *publicKeys {
|
||||
func newPublicKeys(c *Configuration, errorKernel *errorKernel) *publicKeys {
|
||||
p := publicKeys{
|
||||
keysAndHash: newKeysAndHash(),
|
||||
filePath: filepath.Join(c.DatabaseFolder, "publickeys.txt"),
|
||||
keysAndHash: newKeysAndHash(),
|
||||
filePath: filepath.Join(c.DatabaseFolder, "publickeys.txt"),
|
||||
errorKernel: errorKernel,
|
||||
configuration: c,
|
||||
}
|
||||
|
||||
err := p.loadFromFile()
|
||||
|
|
@ -227,7 +237,8 @@ func (p *publicKeys) loadFromFile() error {
|
|||
return err
|
||||
}
|
||||
|
||||
log.Printf("\n ***** DEBUG: Loaded existing keys from file: %v\n\n", p.keysAndHash.Hash)
|
||||
er := fmt.Errorf("nodeAuth: loadFromFile: Loaded existing keys from file: %v", p.keysAndHash.Hash)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
140
processes.go
140
processes.go
|
|
@ -96,28 +96,32 @@ func (p *processes) Start(proc process) {
|
|||
// --- Subscriber services that can be started via flags
|
||||
|
||||
{
|
||||
log.Printf("Starting REQOpProcessList subscriber: %#v\n", proc.node)
|
||||
er := fmt.Errorf("tarting REQOpProcessList subscriber: %#v", proc.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQOpProcessList, string(proc.node))
|
||||
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
{
|
||||
log.Printf("Starting REQOpProcessStart subscriber: %#v\n", proc.node)
|
||||
er := fmt.Errorf("starting REQOpProcessStart subscriber: %#v", proc.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQOpProcessStart, string(proc.node))
|
||||
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
{
|
||||
log.Printf("Starting REQOpProcessStop subscriber: %#v\n", proc.node)
|
||||
er := fmt.Errorf("starting REQOpProcessStop subscriber: %#v", proc.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQOpProcessStop, string(proc.node))
|
||||
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
{
|
||||
log.Printf("Starting REQTest subscriber: %#v\n", proc.node)
|
||||
er := fmt.Errorf("starting REQTest subscriber: %#v", proc.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQTest, string(proc.node))
|
||||
proc := newProcess(proc.ctx, p.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
|
|
@ -259,7 +263,8 @@ func newStartup(server *server) *startup {
|
|||
|
||||
func (s startup) subREQHttpGet(p process) {
|
||||
|
||||
log.Printf("Starting Http Get subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Http Get subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQHttpGet, string(p.node))
|
||||
proc := newProcess(p.ctx, p.processes.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -269,7 +274,9 @@ func (s startup) subREQHttpGet(p process) {
|
|||
|
||||
func (s startup) subREQHttpGetScheduled(p process) {
|
||||
|
||||
log.Printf("Starting Http Get Scheduled subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Http Get Scheduled subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
sub := newSubject(REQHttpGetScheduled, string(p.node))
|
||||
proc := newProcess(p.ctx, p.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -278,7 +285,8 @@ func (s startup) subREQHttpGetScheduled(p process) {
|
|||
}
|
||||
|
||||
func (s startup) pubREQHello(p process) {
|
||||
log.Printf("Starting Hello Publisher: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Hello Publisher: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
sub := newSubject(REQHello, p.configuration.CentralNodeName)
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
|
||||
|
|
@ -307,8 +315,8 @@ func (s startup) pubREQHello(p process) {
|
|||
sam, err := newSubjectAndMessage(m)
|
||||
if err != nil {
|
||||
// In theory the system should drop the message before it reaches here.
|
||||
p.errorKernel.errSend(p, m, err, logError)
|
||||
log.Printf("error: ProcessesStart: %v\n", err)
|
||||
er := fmt.Errorf("error: ProcessesStart: %v", err)
|
||||
p.errorKernel.errSend(p, m, er, logError)
|
||||
}
|
||||
proc.toRingbufferCh <- []subjectAndMessage{sam}
|
||||
|
||||
|
|
@ -317,7 +325,7 @@ func (s startup) pubREQHello(p process) {
|
|||
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)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -329,7 +337,8 @@ func (s startup) pubREQHello(p process) {
|
|||
// to central server and ask for publics keys, and to get them deliver back with a request
|
||||
// of type pubREQKeysDeliverUpdate.
|
||||
func (s startup) pubREQKeysRequestUpdate(p process) {
|
||||
log.Printf("Starting PublicKeysGet Publisher: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting PublicKeysGet Publisher: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
sub := newSubject(REQKeysRequestUpdate, p.configuration.CentralNodeName)
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
|
||||
|
|
@ -365,7 +374,6 @@ func (s startup) pubREQKeysRequestUpdate(p process) {
|
|||
if err != nil {
|
||||
// In theory the system should drop the message before it reaches here.
|
||||
p.errorKernel.errSend(p, m, err, logError)
|
||||
log.Printf("error: ProcessesStart: %v\n", err)
|
||||
}
|
||||
proc.toRingbufferCh <- []subjectAndMessage{sam}
|
||||
|
||||
|
|
@ -374,7 +382,7 @@ func (s startup) pubREQKeysRequestUpdate(p process) {
|
|||
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)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -386,7 +394,8 @@ func (s startup) pubREQKeysRequestUpdate(p process) {
|
|||
// to central server and ask for publics keys, and to get them deliver back with a request
|
||||
// of type pubREQKeysDeliverUpdate.
|
||||
func (s startup) pubREQAclRequestUpdate(p process) {
|
||||
log.Printf("Starting REQAclRequestUpdate Publisher: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting REQAclRequestUpdate Publisher: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
sub := newSubject(REQAclRequestUpdate, p.configuration.CentralNodeName)
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindPublisher, nil)
|
||||
|
|
@ -431,7 +440,7 @@ func (s startup) pubREQAclRequestUpdate(p process) {
|
|||
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)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
|
@ -440,42 +449,48 @@ func (s startup) pubREQAclRequestUpdate(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQKeysRequestUpdate(p process) {
|
||||
log.Printf("Starting Public keys request update subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Public keys request update subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQKeysRequestUpdate, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQKeysDeliverUpdate(p process) {
|
||||
log.Printf("Starting Public keys to Node subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Public keys to Node subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQKeysDeliverUpdate, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQKeysAllow(p process) {
|
||||
log.Printf("Starting Public keys allow subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Public keys allow subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQKeysAllow, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQKeysDelete(p process) {
|
||||
log.Printf("Starting Public keys delete subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Public keys delete subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQKeysDelete, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclRequestUpdate(p process) {
|
||||
log.Printf("Starting Acl Request update subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Request update subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclRequestUpdate, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclDeliverUpdate(p process) {
|
||||
log.Printf("Starting Acl deliver update subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl deliver update subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclDeliverUpdate, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
|
|
@ -484,119 +499,136 @@ func (s startup) subREQAclDeliverUpdate(p process) {
|
|||
// HERE!
|
||||
|
||||
func (s startup) subREQAclAddCommand(p process) {
|
||||
log.Printf("Starting Acl Add Command subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Add Command subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclAddCommand, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclDeleteCommand(p process) {
|
||||
log.Printf("Starting Acl Delete Command subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Delete Command subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclDeleteCommand, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclDeleteSource(p process) {
|
||||
log.Printf("Starting Acl Delete Source subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Delete Source subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclDeleteSource, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupNodesAddNode(p process) {
|
||||
log.Printf("Starting Acl Add node to nodeGroup subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Add node to nodeGroup subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupNodesAddNode, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupNodesDeleteNode(p process) {
|
||||
log.Printf("Starting Acl Delete node from nodeGroup subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Delete node from nodeGroup subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupNodesDeleteNode, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupNodesDeleteGroup(p process) {
|
||||
log.Printf("Starting Acl Delete nodeGroup subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl Delete nodeGroup subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupNodesDeleteGroup, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupCommandsAddCommand(p process) {
|
||||
log.Printf("Starting Acl add command to command group subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl add command to command group subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupCommandsAddCommand, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupCommandsDeleteCommand(p process) {
|
||||
log.Printf("Starting Acl delete command from command group subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl delete command from command group subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupCommandsDeleteCommand, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclGroupCommandsDeleteGroup(p process) {
|
||||
log.Printf("Starting Acl delete command group subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl delete command group subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclGroupCommandsDeleteGroup, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclExport(p process) {
|
||||
log.Printf("Starting Acl export subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl export subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclExport, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQAclImport(p process) {
|
||||
log.Printf("Starting Acl import subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Acl import subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQAclImport, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQToConsole(p process) {
|
||||
log.Printf("Starting Text To Console subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Text To Console subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQToConsole, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQTuiToConsole(p process) {
|
||||
log.Printf("Starting Tui To Console subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Tui To Console subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQTuiToConsole, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQCliCommand(p process) {
|
||||
log.Printf("Starting CLICommand Request subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting CLICommand Request subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQCliCommand, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQPong(p process) {
|
||||
log.Printf("Starting Pong subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Pong subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQPong, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQPing(p process) {
|
||||
log.Printf("Starting Ping Request subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Ping Request subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQPing, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
}
|
||||
|
||||
func (s startup) subREQErrorLog(p process) {
|
||||
log.Printf("Starting REQErrorLog subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting REQErrorLog subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQErrorLog, "errorCentral")
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
go proc.spawnWorker()
|
||||
|
|
@ -610,7 +642,8 @@ func (s startup) subREQErrorLog(p process) {
|
|||
// proc.procFuncCh, and we can then read that message from the procFuncCh in
|
||||
// the procFunc running.
|
||||
func (s startup) subREQHello(p process) {
|
||||
log.Printf("Starting Hello subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting Hello subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQHello, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -630,7 +663,7 @@ func (s startup) subREQHello(p process) {
|
|||
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)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -650,7 +683,8 @@ func (s startup) subREQHello(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQToFile(p process) {
|
||||
log.Printf("Starting text to file subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting text to file subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQToFile, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -658,7 +692,8 @@ func (s startup) subREQToFile(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQToFileNACK(p process) {
|
||||
log.Printf("Starting text to file subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting text to file subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQToFileNACK, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -666,7 +701,8 @@ func (s startup) subREQToFileNACK(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQCopySrc(p process) {
|
||||
log.Printf("Starting copy src subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting copy src subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQCopySrc, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -674,7 +710,8 @@ func (s startup) subREQCopySrc(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQCopyDst(p process) {
|
||||
log.Printf("Starting copy dst subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting copy dst subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQCopyDst, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -682,7 +719,8 @@ func (s startup) subREQCopyDst(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQToFileAppend(p process) {
|
||||
log.Printf("Starting text logging subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting text logging subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQToFileAppend, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -690,7 +728,8 @@ func (s startup) subREQToFileAppend(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQTailFile(p process) {
|
||||
log.Printf("Starting tail log files subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting tail log files subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQTailFile, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -698,7 +737,8 @@ func (s startup) subREQTailFile(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQCliCommandCont(p process) {
|
||||
log.Printf("Starting cli command with continous delivery: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting cli command with continous delivery: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQCliCommandCont, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -706,7 +746,8 @@ func (s startup) subREQCliCommandCont(p process) {
|
|||
}
|
||||
|
||||
func (s startup) subREQPublicKey(p process) {
|
||||
log.Printf("Starting get Public Key subscriber: %#v\n", p.node)
|
||||
er := fmt.Errorf("starting get Public Key subscriber: %#v", p.node)
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
sub := newSubject(REQPublicKey, string(p.node))
|
||||
proc := newProcess(p.ctx, s.server, sub, processKindSubscriber, nil)
|
||||
|
||||
|
|
@ -717,7 +758,8 @@ func (s startup) subREQPublicKey(p process) {
|
|||
|
||||
// Print the content of the processes map.
|
||||
func (p *processes) printProcessesMap() {
|
||||
log.Printf("*** Output of processes map :\n")
|
||||
er := fmt.Errorf("output of processes map : ")
|
||||
p.errorKernel.logConsoleOnlyIfDebug(er, p.configuration)
|
||||
|
||||
{
|
||||
p.active.mu.Lock()
|
||||
|
|
|
|||
Loading…
Reference in a new issue