From 6a7fe51f6009d0b808f3b1436160eb28dd3d6d86 Mon Sep 17 00:00:00 2001 From: postmannen Date: Thu, 19 May 2022 21:35:14 +0200 Subject: [PATCH] added initial REQ for aclDeleteSource --- processes.go | 8 ++++++ requests.go | 5 ++++ requests_acl.go | 74 +++++++++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 85 insertions(+), 2 deletions(-) diff --git a/processes.go b/processes.go index 839413f..fc2ab89 100644 --- a/processes.go +++ b/processes.go @@ -175,6 +175,7 @@ func (p *processes) Start(proc process) { proc.startup.subREQPublicKeysAllow(proc) proc.startup.subREQAclAddCommand(proc) proc.startup.subREQAclDeleteCommand(proc) + proc.startup.subREQAclDeleteSource(proc) } if proc.configuration.StartSubREQPublicKeysToNode { @@ -391,6 +392,13 @@ func (s startup) subREQAclDeleteCommand(p process) { go proc.spawnWorker() } +func (s startup) subREQAclDeleteSource(p process) { + log.Printf("Starting Acl Delete Source subscriber: %#v\n", p.node) + sub := newSubject(REQAclDeleteSource, 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 f732589..51615e7 100644 --- a/requests.go +++ b/requests.go @@ -130,6 +130,8 @@ const ( REQAclAddCommand = "REQAclAddCommand" // REQAclDeleteCommand REQAclDeleteCommand = "REQAclDeleteCommand" + // REQAclDeleteSource + REQAclDeleteSource = "REQAclDeleteSource" ) // The mapping of all the method constants specified, what type @@ -226,6 +228,9 @@ func (m Method) GetMethodsAvailable() MethodsAvailable { REQAclDeleteCommand: methodREQAclDeleteCommand{ event: EventACK, }, + REQAclDeleteSource: methodREQAclDeleteSource{ + event: EventACK, + }, }, } diff --git a/requests_acl.go b/requests_acl.go index f13a28b..a061629 100644 --- a/requests_acl.go +++ b/requests_acl.go @@ -95,7 +95,7 @@ func (m methodREQAclDeleteCommand) handler(proc process, message Message, node s switch { case len(message.MethodArgs) < 3: - er := fmt.Errorf("error: methodREQAclAddAccessList: got <3 number methodArgs, want 3") + er := fmt.Errorf("error: methodREQAclDeleteCommand: got <3 number methodArgs, want 3") proc.errorKernel.errSend(proc, message, er) return @@ -130,7 +130,77 @@ func (m methodREQAclDeleteCommand) handler(proc process, message Message, node s case <-ctx.Done(): cancel() - er := fmt.Errorf("error: methodREQAclAddAccessList: method timed out: %v", message.MethodArgs) + er := fmt.Errorf("error: methodREQAclDeleteCommand: method timed out: %v", message.MethodArgs) + 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 +} + +// --- + +type methodREQAclDeleteSource struct { + event Event +} + +func (m methodREQAclDeleteSource) getKind() Event { + return m.event +} + +func (m methodREQAclDeleteSource) handler(proc process, message Message, node string) ([]byte, error) { + inf := fmt.Errorf("<--- methodREQAclDeleteSource 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() + + switch { + case len(message.MethodArgs) < 2: + er := fmt.Errorf("error: methodREQAclDeleteSource: got <2 number methodArgs, want 2") + proc.errorKernel.errSend(proc, message, er) + + return + } + + // Get a context with the timeout specified in message.MethodTimeout. + ctx, cancel := getContextForMethodTimeout(proc.ctx, message) + + outCh := make(chan []byte) + + proc.processes.wg.Add(1) + go func() { + defer proc.processes.wg.Done() + + host := message.MethodArgs[0] + source := message.MethodArgs[1] + + proc.centralAuth.accessLists.aclDeleteSource(Node(host), Node(source)) + + outString := fmt.Sprintf("acl deleted: host=%v, source=%v\n", host, source) + out := []byte(outString) + + select { + case outCh <- out: + case <-ctx.Done(): + return + } + }() + + select { + case <-ctx.Done(): + + cancel() + er := fmt.Errorf("error: methodREQAclDeleteSource: method timed out: %v", message.MethodArgs) proc.errorKernel.errSend(proc, message, er) case out := <-outCh: