<commit_before>package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"

	"golang.org/x/net/websocket"
)

var token string

func init() {
	if token = os.Getenv("TOKEN"); token == "" {
		log.Fatal("TOKEN not specified")
	}
}

//{"type":"message","channel":"D04FLBCPZ","user":"U02D58RR5","text":"howdy ho","ts":"1429561458.000015","team":"T02D58RR3"}
type Message struct {
	Type    string
	Channel string
	User    string
	Text    string
	Ts      string
	Team    string
}

func webSocketUrl() (url string, err error) {
	resp, err := http.Get("https://slack.com/api/rtm.start?token=" + token)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	buf, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	var body map[string]interface{}
	if err = json.Unmarshal(buf, &body); err != nil {
		return
	}
	url, ok := body["url"].(string)
	if !ok {
		err = errors.New("unable to parse rtm url")
	}
	return
}

func main() {
	url, err := webSocketUrl()
	if err != nil {
		log.Fatal(err)
	}
	origin := "http://localhost/"
	ws, err := websocket.Dial(url, "", origin)
	if err != nil {
		log.Fatal(err)
	}
	defer ws.Close()

	input := make(chan []byte)
	go func() {
		for {
			var buf []byte
			if err := websocket.Message.Receive(ws, &buf); err != nil {
				log.Println(ws)
				log.Fatal(err)
			}
			if len(buf) > 0 {
				input <- buf
			}
		}
	}()

	output := make(chan map[string]interface{})
	go func() {
		id := 0
		for {
			msg := <-output
			msg["id"] = id
			buf, _ := json.MarshalIndent(msg, "", "  ")
			log.Println("sending message")
			fmt.Println(string(buf))
			if err := websocket.JSON.Send(ws, msg); err != nil {
				log.Fatal(err)
			}
			id++
		}
	}()

	messages := make(chan Message)
	go func() {
		for {
			select {
			case buf := <-input:
				var msg Message
				if err := json.Unmarshal(buf, &msg); err != nil {
					log.Fatal("unable to parse rtm message")
				}
				switch msg.Type {
				case "message":
					if len(msg.Channel) > 0 && msg.Channel[0] == 'D' {
						messages <- msg
					}
				default:
					log.Println("ignoring", msg.Type)
				}
			case <-time.After(5 * time.Second):
				log.Println("timeout, pinging")
				output <- map[string]interface{}{"type": "ping"}
			}
		}
	}()
	for m := range messages {
		fmt.Println(m)
		output <- map[string]interface{}{
			"type":    m.Type,
			"channel": m.Channel,
			"text":    m.Text,
		}
	}
}<commit_msg>refactor and add reconnecting<commit_after>package main

import (
	"encoding/json"
	"errors"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"

	"golang.org/x/net/websocket"
)

type Message map[string]interface{}

func (m Message) Type() string    { return m["type"].(string) }
func (m Message) Channel() string { return m["channel"].(string) }
func (m Message) User() string    { return m["user"].(string) }
func (m Message) Text() string    { return m["text"].(string) }

func (m Message) DirectMessage() bool {
	if _, ok := m["reply_to"]; ok {
		return false
	}
	msgType, ok := m["type"].(string)
	if !ok {
		return false
	}
	channel, ok := m["channel"].(string)
	if !ok {
		return false
	}
	if _, ok := m["user"].(string); !ok {
		return false
	}
	if _, ok := m["text"].(string); !ok {
		return false
	}
	return msgType == "message" && len(channel) > 0 && channel[0] == 'D'
}

func (m Message) String() string {
	buf, _ := json.Marshal(m)
	return string(buf)
}

var token string

func init() {
	if token = os.Getenv("TOKEN"); token == "" {
		log.Fatal("TOKEN not specified")
	}
}

func webSocketUrl() (url string, err error) {
	log.Println("authenticating with rtm")
	resp, err := http.Get("https://slack.com/api/rtm.start?token=" + token)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	buf, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return
	}
	var body map[string]interface{}
	if err = json.Unmarshal(buf, &body); err != nil {
		return
	}
	url, ok := body["url"].(string)
	if !ok {
		err = errors.New("unable to parse rtm url")
	}
	return
}

func connectToRtm() (ws *websocket.Conn, input, output chan Message, err error) {
	var url string
	url, err = webSocketUrl()
	if err != nil {
		return
	}

	log.Println("connecting to rtm")
	origin := "http://localhost/"
	ws, err = websocket.Dial(url, "", origin)
	if err != nil {
		return
	}

	input = make(chan Message)
	go func() {
		for {
			var msg Message
			if err := websocket.JSON.Receive(ws, &msg); err != nil {
				close(input)
				return
			}
			if msg != nil {
				input <- msg
			}
		}
	}()

	output = make(chan Message)
	go func() {
		id := 0
		for {
			msg := <-output
			msg["id"] = id
			if err := websocket.JSON.Send(ws, msg); err != nil {
				close(output)
				return
			}
			id++
		}
	}()

	return
}

func main() {
	ws, input, output, err := connectToRtm()
	if err != nil {
		log.Fatal(err)
	}
	defer ws.Close()

	messages := make(chan Message)
	go func() {
		for {
			select {
			case msg, ok := <-input:
				// Handle disconects.
				if !ok {
					log.Println("disconnected from rtm")
					ws, input, output, err = connectToRtm()
					if err != nil {
						log.Fatal(err)
					}
					defer ws.Close()
					continue
				}
				if msg.DirectMessage() {
					messages <- msg
				}
			case <-time.After(5 * time.Second):
				output <- Message{"type": "ping"}
			}
		}
	}()

	for m := range messages {
		log.Println(m)
		output <- Message{
			"type":    m.Type(),
			"channel": m.Channel(),
			"text":    m.Text(),
		}
	}
}
{'fraction_non_alphanumeric': 0.18012536873156343, 'fraction_numerical': 0.006821533923303835, 'mean_word_length': 2.850248403122782, 'pattern_counts': {'":': 15, '<': 17, '<?xml version=': 0, '>': 6, 'https://': 2, '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': '2023352', 'n_tokens_mistral': 2370, 'n_tokens_neox': 2061, 'n_words': 514}