diff --git a/processes.go b/processes.go index 05860c7..8affffe 100644 --- a/processes.go +++ b/processes.go @@ -183,6 +183,7 @@ func (p *processes) Start(proc process) { proc.startup.subREQAclGroupCommandsDeleteCommand(proc) proc.startup.subREQAclGroupCommandsDeleteGroup(proc) proc.startup.subREQAclExport(proc) + proc.startup.subREQAclImport(proc) } if proc.configuration.StartSubREQPublicKeysToNode { @@ -455,6 +456,13 @@ func (s startup) subREQAclExport(p process) { go proc.spawnWorker() } +func (s startup) subREQAclImport(p process) { + log.Printf("Starting Acl import subscriber: %#v\n", p.node) + 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) sub := newSubject(REQToConsole, string(p.node)) diff --git a/requests.go b/requests.go index 7fcf395..5a0e2a7 100644 --- a/requests.go +++ b/requests.go @@ -146,6 +146,8 @@ const ( REQAclGroupCommandsDeleteGroup = "REQAclGroupCommandsDeleteGroup" // REQAclExport REQAclExport = "REQAclExport" + // REQAclImport + REQAclImport = "REQAclImport" ) // The mapping of all the method constants specified, what type @@ -271,6 +273,9 @@ func (m Method) GetMethodsAvailable() MethodsAvailable { REQAclExport: methodREQAclExport{ event: EventACK, }, + REQAclImport: methodREQAclImport{ + event: EventACK, + }, }, } diff --git a/requests_acl.go b/requests_acl.go index aab2eac..df77f6b 100644 --- a/requests_acl.go +++ b/requests_acl.go @@ -706,3 +706,75 @@ func (m methodREQAclExport) handler(proc process, message Message, node string) ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID)) return ackMsg, nil } + +// --- + +type methodREQAclImport struct { + event Event +} + +func (m methodREQAclImport) getKind() Event { + return m.event +} + +func (m methodREQAclImport) handler(proc process, message Message, node string) ([]byte, error) { + inf := fmt.Errorf("<--- methodREQAclImport received from: %v, containing: %v", message.FromNode, message.MethodArgs) + proc.errorKernel.logConsoleOnlyIfDebug(inf, proc.configuration) + + proc.processes.wg.Add(1) + go func() { + defer proc.processes.wg.Done() + + // Get a context with the timeout specified in message.MethodTimeout. + ctx, cancel := getContextForMethodTimeout(proc.ctx, message) + + outCh := make(chan []byte) + errCh := make(chan error) + + switch { + case len(message.MethodArgs) < 1: + errCh <- fmt.Errorf("error: methodREQAclImport: got <1 number methodArgs, want 1") + return + } + + proc.processes.wg.Add(1) + go func() { + defer proc.processes.wg.Done() + + js := []byte(message.MethodArgs[0]) + err := proc.centralAuth.accessLists.importACLs(js) + if err != nil { + errCh <- fmt.Errorf("error: methodREQAclImport failed: %v", err) + return + } + + outString := fmt.Sprintf("Imported acl's sent from: %v\n", message.FromNode) + out := []byte(outString) + + select { + case outCh <- out: + case <-ctx.Done(): + return + } + }() + + select { + case err := <-errCh: + proc.errorKernel.errSend(proc, message, err) + + case <-ctx.Done(): + cancel() + er := fmt.Errorf("error: methodREQAclImport: method timed out") + proc.errorKernel.errSend(proc, message, er) + + case out := <-outCh: + // Prepare and queue for sending a new message with the output + // of the action executed. + newReplyMessage(proc, message, out) + } + + }() + + ackMsg := []byte("confirmed from: " + node + ": " + fmt.Sprint(message.ID)) + return ackMsg, nil +}