<commit_before>package rtm

import (
	"context"
	"encoding/json"
	"log"
	"time"

	"github.com/cenkalti/backoff"
	"github.com/gorilla/websocket"
	"github.com/lestrrat/go-slack"
	"github.com/pkg/errors"
)

func New(cl *slack.Client) *Client {
	return &Client{
		client:       cl,
		eventsCh:     make(chan *Event),
	}
}

func (c *Client) Events() <-chan *Event {
	return c.eventsCh
}

// Run starts the RTM run loop.
func (c *Client) Run(octx context.Context) error {
	octxwc, cancel := context.WithCancel(octx)
	defer cancel()

	ctx := newRtmCtx(octxwc, c.eventsCh)
	go ctx.run()

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		ctx.emit(&Event{typ: ClientConnectingEventType})

		var conn *websocket.Conn

		strategy := backoff.NewExponentialBackOff()
		strategy.InitialInterval = 100 * time.Millisecond
		strategy.MaxInterval = 5 * time.Second
		strategy.MaxElapsedTime = 0
		err := backoff.Retry(func() error {
			res, err := c.client.RTM().Start().Do(ctx)
			if err != nil {
				log.Printf("failed to start RTM sesson: %s", err)
				return err
			}
			conn, _, err = websocket.DefaultDialer.Dial(res.URL, nil)
			if err != nil {
				log.Printf("failed to dial to websocket: %s", err)
				return err
			}
			return nil
		}, backoff.WithContext(strategy, ctx))

		if err != nil {
			return errors.Wrap(err, `failed to connect to RTM endpoint`)
		}

		ctx.handleConn(conn)
		// we get here if we manually canceled the context
		// of if the websocket ReadMessage returned an error
		ctx.emit(&Event{typ: ClientDisconnectedEventType})

	}

	return nil
}

func (ctx *rtmCtx) handleConn(conn *websocket.Conn) {
	defer conn.Close()

	in := make(chan []byte)

	// This goroutine is responsible for reading from the
	// websocket connection. It's separated because the
	// ReadMessage() operation is blocking.
	go func(ch chan []byte, conn *websocket.Conn) {
		defer close(ch)

		for {
			typ, data, err := conn.ReadMessage()
			if err != nil {
				// There was an error. we need to bail out
				return
			}
			// we only understand text messages
			if typ != websocket.TextMessage {
				continue
			}
			ch <- data
		}
	}(in, conn)

	for {
		select {
		case <-ctx.Done():
			return
		case payload, ok := <-in:
			if !ok {
				// if the channel is closed, we probably had some
				// problems in the ReadMessage proxy. bail out
				return
			}
			log.Printf("raw payload: %s", payload)
			var event Event
			if err := json.Unmarshal(payload, &event); err != nil {
				log.Printf("failed to unmarshal: %s", err)
			}
		}
	}
}

type rtmCtx struct {
	context.Context
	inbuf        chan *Event
	outbuf       chan<- *Event
	writeTimeout time.Duration
}

func newRtmCtx(octx context.Context, outch chan<- *Event) *rtmCtx {
	return &rtmCtx{
		Context:      octx,
		inbuf:        make(chan *Event),
		outbuf:       outch,
		writeTimeout: 500 * time.Millisecond,
	}<commit_msg>whoa, forgot to proxy the event<commit_after>package rtm

import (
	"context"
	"encoding/json"
	"log"
	"time"

	"github.com/cenkalti/backoff"
	"github.com/gorilla/websocket"
	pdebug "github.com/lestrrat/go-pdebug"
	"github.com/lestrrat/go-slack"
	"github.com/pkg/errors"
)

func New(cl *slack.Client) *Client {
	return &Client{
		client:   cl,
		eventsCh: make(chan *Event),
	}
}

func (c *Client) Events() <-chan *Event {
	return c.eventsCh
}

// Run starts the RTM run loop.
func (c *Client) Run(octx context.Context) error {
	octxwc, cancel := context.WithCancel(octx)
	defer cancel()

	ctx := newRtmCtx(octxwc, c.eventsCh)
	go ctx.run()

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}

		ctx.emit(&Event{typ: ClientConnectingEventType})

		var conn *websocket.Conn

		strategy := backoff.NewExponentialBackOff()
		strategy.InitialInterval = 100 * time.Millisecond
		strategy.MaxInterval = 5 * time.Second
		strategy.MaxElapsedTime = 0
		err := backoff.Retry(func() error {
			res, err := c.client.RTM().Start().Do(ctx)
			if err != nil {
				log.Printf("failed to start RTM sesson: %s", err)
				return err
			}
			conn, _, err = websocket.DefaultDialer.Dial(res.URL, nil)
			if err != nil {
				log.Printf("failed to dial to websocket: %s", err)
				return err
			}
			return nil
		}, backoff.WithContext(strategy, ctx))

		if err != nil {
			return errors.Wrap(err, `failed to connect to RTM endpoint`)
		}

		ctx.handleConn(conn)
		// we get here if we manually canceled the context
		// of if the websocket ReadMessage returned an error
		ctx.emit(&Event{typ: ClientDisconnectedEventType})

	}

	return nil
}

func (ctx *rtmCtx) handleConn(conn *websocket.Conn) {
	defer conn.Close()

	in := make(chan []byte)

	// This goroutine is responsible for reading from the
	// websocket connection. It's separated because the
	// ReadMessage() operation is blocking.
	go func(ch chan []byte, conn *websocket.Conn) {
		defer close(ch)

		for {
			typ, data, err := conn.ReadMessage()
			if err != nil {
				// There was an error. we need to bail out
				if pdebug.Enabled {
					pdebug.Printf("error while reading message from websocket: %s", err)
				}
				return
			}

			// we only understand text messages
			if typ != websocket.TextMessage {
				if pdebug.Enabled {
					pdebug.Printf("received websocket message, but it is not a text payload. refusing to process")
				}
				continue
			}
			if pdebug.Enabled {
				pdebug.Printf("forwarding new websocket message")
			}
			ch <- data
		}
	}(in, conn)

	for {
		select {
		case <-ctx.Done():
			return
		case payload, ok := <-in:
			if !ok {
				if pdebug.Enabled {
					pdebug.Printf("websocket proxy: detected incoming channel close.")
				}
				// if the channel is closed, we probably had some
				// problems in the ReadMessage proxy. bail out
				return
			}

			if pdebug.Enabled {
				pdebug.Printf("websocket proxy: received raw payload: %s", payload)
			}

			var event Event
			if err := json.Unmarshal(payload, &event); err != nil {
				if pdebug.Enabled {
					pdebug.Printf("websocket proxy: failed to unmarshal payload: %s", err)
				}
			}

			ctx.inbuf <- &event
		}
	}
}

type rtmCtx struct {
	context.Context
	inbuf        chan *Event
	outbuf       chan<- *Event
	writeTimeout time.Duration
}

func newRtmCtx(octx context.Context, outch chan<- *Event) *rtmCtx {
	return &rtmCtx{
		Context:      octx,
		inbuf:        make(chan *Event),
		outbuf:       outch,
		writeTimeout: 500 * time.Millisecond,
	}
{'fraction_non_alphanumeric': 0.12315888436226888, 'fraction_numerical': 0.0025070510811657787, 'mean_word_length': 2.991869918699187, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1299481', 'n_tokens_mistral': 2544, 'n_tokens_neox': 2200, 'n_words': 636}