summaryrefslogtreecommitdiff
path: root/internal/clients/connectors/serverless.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/clients/connectors/serverless.go')
-rw-r--r--internal/clients/connectors/serverless.go90
1 files changed, 90 insertions, 0 deletions
diff --git a/internal/clients/connectors/serverless.go b/internal/clients/connectors/serverless.go
new file mode 100644
index 0000000..0500645
--- /dev/null
+++ b/internal/clients/connectors/serverless.go
@@ -0,0 +1,90 @@
+package connectors
+
+import (
+ "context"
+ "io"
+
+ "github.com/mimecast/dtail/internal/clients/handlers"
+ "github.com/mimecast/dtail/internal/config"
+ "github.com/mimecast/dtail/internal/io/logger"
+ serverHandlers "github.com/mimecast/dtail/internal/server/handlers"
+ user "github.com/mimecast/dtail/internal/user/server"
+)
+
+// Serverless creates a server object directly without TCP.
+type Serverless struct {
+ handler handlers.Handler
+ commands []string
+ userName string
+}
+
+// NewServerConnection returns a new connection.
+func NewServerless(userName string, handler handlers.Handler, commands []string) *Serverless {
+ s := Serverless{
+ userName: userName,
+ handler: handler,
+ commands: commands,
+ }
+
+ logger.Debug("Creating new serverless connector", handler, commands)
+ return &s
+}
+
+func (s *Serverless) Server() string {
+ return "local(serverless)"
+}
+
+func (s *Serverless) Handler() handlers.Handler {
+ return s.handler
+}
+
+func (s *Serverless) Start(ctx context.Context, cancel context.CancelFunc, throttleCh, statsCh chan struct{}) {
+ go func() {
+ defer cancel()
+
+ if err := s.handle(ctx, cancel); err != nil {
+ logger.Warn(err)
+ }
+ }()
+
+ <-ctx.Done()
+}
+
+func (s *Serverless) handle(ctx context.Context, cancel context.CancelFunc) error {
+ logger.Debug("Creating server handler for a serverless session")
+
+ serverHandler := serverHandlers.NewServerHandler(
+ user.New(s.userName, s.Server()),
+ make(chan struct{}, config.Server.MaxConcurrentCats),
+ make(chan struct{}, config.Server.MaxConcurrentTails),
+ )
+
+ go func() {
+ io.Copy(serverHandler, s.handler)
+ cancel()
+ }()
+
+ go func() {
+ io.Copy(s.handler, serverHandler)
+ cancel()
+ }()
+
+ go func() {
+ select {
+ case <-s.handler.Done():
+ case <-ctx.Done():
+ }
+ cancel()
+ }()
+
+ // Send all commands to client.
+ for _, command := range s.commands {
+ logger.Debug(command)
+ s.handler.SendMessage(command)
+ }
+
+ <-ctx.Done()
+ s.handler.Shutdown()
+
+ return nil
+}