summaryrefslogtreecommitdiff
path: root/internal/server/handlers/mapcommand.go
diff options
context:
space:
mode:
authorPaul Bütow <pbuetow@mimecast.com>2020-01-26 11:26:53 +0000
committerPaul Bütow <pbuetow@mimecast.com>2020-02-07 13:31:15 +0000
commit0945da8dfefcbb723eecea0e5f4eafff63398253 (patch)
treef06dab4d2bf21d25d176b23d5baeca588d27f5d7 /internal/server/handlers/mapcommand.go
parent2a8e5de265a0e0a31a5834909d6879f5c9941467 (diff)
Introduce drun command, refactor code to use context package
Diffstat (limited to 'internal/server/handlers/mapcommand.go')
-rw-r--r--internal/server/handlers/mapcommand.go35
1 files changed, 35 insertions, 0 deletions
diff --git a/internal/server/handlers/mapcommand.go b/internal/server/handlers/mapcommand.go
new file mode 100644
index 0000000..10372da
--- /dev/null
+++ b/internal/server/handlers/mapcommand.go
@@ -0,0 +1,35 @@
+package handlers
+
+import (
+ "context"
+ "strings"
+
+ "github.com/mimecast/dtail/internal/mapr/server"
+)
+
+// Map command implements the mapreduce command server side.
+type mapCommand struct {
+ aggregate *server.Aggregate
+ server *ServerHandler
+}
+
+// NewMapCommand returns a new server side mapreduce command.
+func newMapCommand(serverHandler *ServerHandler, argc int, args []string) (mapCommand, *server.Aggregate, error) {
+ mapCommand := mapCommand{
+ server: serverHandler,
+ }
+
+ queryStr := strings.Join(args[1:], " ")
+ aggregate, err := server.NewAggregate(queryStr)
+ if err != nil {
+ return mapCommand, nil, err
+ }
+
+ mapCommand.aggregate = aggregate
+ return mapCommand, aggregate, nil
+
+}
+
+func (m mapCommand) Start(ctx context.Context, aggregatedMessages chan<- string) {
+ m.aggregate.Start(ctx, aggregatedMessages)
+}