From 9280ecb7e0b0039d6c1f4800373eb76452145078 Mon Sep 17 00:00:00 2001 From: René 'Necoro' Neumann Date: Wed, 22 Apr 2020 23:40:53 +0200 Subject: Concurrent feed processing; central imap handler --- internal/imap/commando.go | 92 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 internal/imap/commando.go (limited to 'internal/imap/commando.go') diff --git a/internal/imap/commando.go b/internal/imap/commando.go new file mode 100644 index 0000000..fbfeaec --- /dev/null +++ b/internal/imap/commando.go @@ -0,0 +1,92 @@ +package imap + +import "github.com/Necoro/feed2imap-go/internal/log" + +const maxPipeDepth = 10 + +type commander struct { + client *Client + pipe chan<- execution + done chan<- struct{} +} + +type command interface { + execute(client *Client) error +} + +type ErrorHandler func(error) string + +type execution struct { + cmd command + done chan<- struct{} + errorHandler ErrorHandler +} + +type addCommando struct { + folder Folder + messages []string +} + +func (cmd addCommando) execute(client *Client) error { + return client.putMessages(cmd.folder, cmd.messages) +} + +type ensureCommando struct { + folder Folder +} + +func (cmd ensureCommando) execute(client *Client) error { + return client.ensureFolder(cmd.folder) +} + +func (commander *commander) execute(command command, handler ErrorHandler) { + done := make(chan struct{}) + commander.pipe <- execution{command, done, handler} + <-done +} + +func executioner(client *Client, pipe <-chan execution, done <-chan struct{}) { + for { + select { + case <-done: + return + case execution := <-pipe: + select { // break as soon as done is there + case <-done: + return + default: + } + if err := execution.cmd.execute(client); err != nil { + if execution.errorHandler == nil { + log.Error(err) + } else { + log.Error(execution.errorHandler(err)) + } + } + close(execution.done) + } + } +} + +func (client *Client) startCommander() { + if client.commander != nil { + return + } + + pipe := make(chan execution, maxPipeDepth) + done := make(chan struct{}) + + client.commander = &commander{client, pipe, done} + + go executioner(client, pipe, done) +} + +func (client *Client) stopCommander() { + if client.commander == nil { + return + } + + close(client.commander.done) + + client.commander = nil +} -- cgit v1.2.3-54-g00ecf