rtmapi.go

package slack

import (
	"context"
	"fmt"
	"github.com/oklahomer/go-kasumi/logger"
	"github.com/oklahomer/go-kasumi/retry"
	"github.com/oklahomer/go-sarah/v4"
	"github.com/oklahomer/golack/v2/event"
	"github.com/oklahomer/golack/v2/rtmapi"
	"strings"
	"time"
)

const pingSignalChannelID = "ping"

type rtmAPIAdapter struct {
	config        *Config
	client        SlackClient
	handlePayload func(context.Context, *Config, rtmapi.DecodedPayload, func(sarah.Input) error)
}

var _ apiSpecificAdapter = (*rtmAPIAdapter)(nil)

func (r *rtmAPIAdapter) run(ctx context.Context, enqueueInput func(sarah.Input) error, notifyErr func(error)) {
	for {
		conn, err := r.connect(ctx)
		if err != nil {
			// Failed to establish a WebSocket connection with max retrials.
			// Notify the unrecoverable state and give up.
			notifyErr(sarah.NewBotNonContinuableError(err.Error()))
			return
		}

		// Create a connection specific context so each connection-scoped goroutine can receive the connection closing signal and eventually return.
		connCtx, connCancel := context.WithCancel(ctx)

		// This channel is not subject to close. This channel can be accessed in a parallel manner with nonBlockSignal function,
		// and the receiver is NOT waiting for a close signal. Let GC run when this channel is no longer referred.
		//
		// http://stackoverflow.com/a/8593986
		// "Note that it is only necessary to close a channel if the receiver is looking for a close.
		// Closing the channel is a control signal on the channel indicating that no more data follows."
		tryPing := make(chan struct{}, 1)

		go r.receivePayload(connCtx, conn, tryPing, enqueueInput)

		// Payload reception and other connection-related tasks must run in separate goroutines since receivePayload function
		// internally blocks till the per-connection context is cancelled.
		connErr := r.superviseConnection(connCtx, conn, tryPing)

		// superviseConnection returns when parent context is canceled or the connection is hopelessly unstable.
		// Close the current connection and do some cleanup.
		_ = conn.Close()
		connCancel()
		if connErr == nil {
			// Connection is intentionally closed by the caller.
			// No more interaction follows.
			return
		}

		logger.Errorf("Will try re-connection due to previous connection's fatal state: %+v", connErr)
	}
}

func (r *rtmAPIAdapter) connect(ctx context.Context) (rtmapi.Connection, error) {
	var conn rtmapi.Connection
	err := retry.WithPolicy(r.config.RetryPolicy, func() (e error) {
		conn, e = r.client.ConnectRTM(ctx)
		return e
	})
	return conn, err
}

func (r *rtmAPIAdapter) receivePayload(connCtx context.Context, payloadReceiver rtmapi.PayloadReceiver, tryPing chan<- struct{}, enqueueInput func(sarah.Input) error) {
	for {
		select {
		case <-connCtx.Done():
			logger.Info("Stop receiving payload due to context cancel")
			return

		default:
			payload, err := payloadReceiver.Receive()
			// TODO should io.EOF and io.ErrUnexpectedEOF treated differently than other errors?
			if err == event.ErrEmptyPayload {
				continue
			}

			switch err.(type) {
			case nil:
				// O.K. Do nothing and proceed to the payload handling

			case *event.MalformedPayloadError:
				logger.Warnf("Ignore malformed payload: %+v", err)
				continue

			case *rtmapi.UnexpectedMessageTypeError:
				logger.Warnf("Ignore a payload with unexpected message type: %+v", err)
				continue

			default:
				// Connection might not be stable or is closed already.
				logger.Infof("Try ping caused by error: %+v", err)
				nonBlockSignal(pingSignalChannelID, tryPing)
				continue
			}

			if payload == nil {
				continue
			}

			r.handlePayload(connCtx, r.config, payload, enqueueInput)
		}
	}
}

func (r *rtmAPIAdapter) superviseConnection(connCtx context.Context, payloadSender rtmapi.PayloadSender, tryPing chan struct{}) error {
	ticker := time.NewTicker(r.config.PingInterval)
	defer ticker.Stop()

	for {
		select {
		case <-connCtx.Done():
			return nil

		case <-ticker.C:
			nonBlockSignal(pingSignalChannelID, tryPing)

		case <-tryPing:
			logger.Debug("Send ping")
			err := payloadSender.Ping()
			if err != nil {
				return fmt.Errorf("error on ping: %w", err)
			}
		}
	}
}

// DefaultRTMPayloadHandler receives incoming events, converts them to sarah.Input, and then passes them to enqueueInput.
// To replace this default behavior, define a function with the same signature and replace this.
//
//   myHandler := func(_ context.Context, config *Config, _ rtmapi.DecodedPayload, _ func(sarah.Input) error) {}
//   slackAdapter, _ := slack.NewAdapter(slackConfig, slack.WithRTMPayloadHandler(myHandler))
func DefaultRTMPayloadHandler(_ context.Context, config *Config, payload rtmapi.DecodedPayload, enqueueInput func(sarah.Input) error) {
	switch p := payload.(type) {
	case *rtmapi.OKReply:
		logger.Debugf("Successfully sent. ID: %d. Text: %s.", p.ReplyTo, p.Text)

	case *rtmapi.NGReply:
		logger.Errorf(
			"Something was wrong with previous message sending. id: %d. error code: %d. error message: %s.",
			p.ReplyTo, p.Error.Code, p.Error.Message)

	case *rtmapi.Pong:
		logger.Debug("Pong message received.")

	case *event.Hello:
		logger.Debugf("Successfully connected.")

	default:
		input, err := EventToInput(p)
		if err == ErrNonSupportedEvent {
			logger.Debugf("Event given, but no corresponding action is defined. %#v", payload)
			return
		}

		if err != nil {
			logger.Errorf("Failed to convert %T event: %s", p, err.Error())
			return
		}

		trimmed := strings.TrimSpace(input.Message())
		if config.HelpCommand != "" && trimmed == config.HelpCommand {
			// Help command
			help := sarah.NewHelpInput(input)
			_ = enqueueInput(help)
		} else if config.AbortCommand != "" && trimmed == config.AbortCommand {
			// Abort command
			abort := sarah.NewAbortInput(input)
			_ = enqueueInput(abort)
		} else {
			// Regular input
			_ = enqueueInput(input)
		}
	}
}