func (cr *consumer) inboxDispatcher() {
	for message := range cr.inbox {
		cr.connection.SetWriteDeadline(time.Now().Add(cr.es.settings.GetTimeout()))
		if _, err := cr.connection.Write(message.Message()); err != nil {
			if netErr, ok := err.(net.Error); !ok || netErr.Timeout() {
				cr.expired = true
				cr.connection.Close()
				cr.es.expireConsumer <- cr
				return
			}
		}
	}
	cr.connection.Close()
}
InboxDispatcher processes incoming eventMessages.
It disconnects timed out consumers and initiates the removal from the consumer pool.
package eventsource
import (
	"bytes"
	"fmt"
	"net"
	"net/http"
	"time"
)
type consumer struct {
	connection net.Conn
	es         *eventSource
	inbox      chan *eventMessage
	channel    string
	expired    bool
}
func newConsumer(resp http.ResponseWriter, req *http.Request, es *eventSource, channel string) (*consumer, error) {
	connection, _, err := resp.(http.Hijacker).Hijack()
	if err != nil {
		return nil, err
	}
	cr := &consumer{
		connection: connection,
		es:         es,
		inbox:      make(chan *eventMessage),
		channel:    channel,
		expired:    false,
	}
	if err := cr.setupConnection(); err != nil {
		return nil, err
	}
	go cr.inboxDispatcher()
	return cr, nil
}
func (cr *consumer) setupConnection() error {
	headers := [][]byte{
		[]byte("HTTP/1.1 200 OK"),
		[]byte("Content-Type: text/event-stream"),
		[]byte("Cache-Control: no-cache"),
		[]byte("Connection: keep-alive"),
		[]byte(fmt.Sprintf("Access-Control-Allow-Origin: %s", cr.es.settings.GetCorsAllowOrigin())),
		[]byte(fmt.Sprintf("Access-Control-Allow-Method: %s", cr.es.settings.GetCorsAllowMethod())),
	}
	headersData := append(bytes.Join(headers, []byte("\n")), []byte("\n\n")...)
	if _, err := cr.connection.Write(headersData); err != nil {
		cr.connection.Close()
		return err
	}
	return nil
}