func validContentType(contentType string) bool {
	if strings.Contains(strings.ToLower(contentType), "application/json") {
		return true
	}
	return false
}
ValidContentType validates the submitted Content-Type.
package eventsource
import (
	"fmt"
	"github.com/gorilla/mux"
	"io"
	"log"
	"net/http"
	"runtime"
	"sort"
	"strings"
)
const (
	globalChannel = "all"
)
type EventSource interface {
	Router() *mux.Router
	SendMessage(io.Reader, string)
	ChannelExists(channel string) bool
	ConsumerCount(channel string) int
	ConsumerCountAll() int
	Channels() []string
	Close(channel string)
	CloseAll()
	Run()
	Stop()
}
type eventSource struct {
	messageRouter   chan *eventMessage
	expireConsumer  chan *consumer
	addConsumer     chan *consumer
	closeChannel    chan string
	stopApplication chan bool
	settings        *Settings
	consumers       map[string][]*consumer
}
func New(settings *Settings) EventSource {
	if settings == nil {
		settings = &Settings{}
	}
	es := &eventSource{
		messageRouter:   make(chan *eventMessage),
		expireConsumer:  make(chan *consumer),
		addConsumer:     make(chan *consumer),
		closeChannel:    make(chan string),
		stopApplication: make(chan bool),
		settings:        settings,
		consumers:       make(map[string][]*consumer),
	}
	go es.actionDispatcher()
	return es
}
func (es *eventSource) Router() *mux.Router {
	router := mux.NewRouter()
	router.HandleFunc("/{channel:[a-z0-9-_]+}", es.subscribeHandler).Methods("GET")
	router.HandleFunc("/{channel:[a-z0-9-_]+}", es.publishHandler).Methods("POST")
	router.HandleFunc("/{channel:[a-z0-9-_]+}", es.closeHandler).Methods("DELETE")
	router.HandleFunc("/{channel:[a-z0-9-_]+}", es.informationHandler).Methods("HEAD")
	router.NotFoundHandler = http.HandlerFunc(channelNotFoundHandler)
	return router
}
func (es *eventSource) SendMessage(messageStream io.Reader, channel string) {
	em, err := newEventMessage(messageStream, channel)
	if err != nil {
		log.Printf("[E] Unable to create event message for channel '%s'. %s", channel, err)
		return
	}
	es.messageRouter <- em
}
func (es *eventSource) ChannelExists(channel string) bool {
	_, ok := es.consumers[channel]
	return ok
}
func (es *eventSource) ConsumerCount(channel string) int {
	if consumers, ok := es.consumers[channel]; ok {
		return len(consumers)
	}
	return 0
}
func (es *eventSource) ConsumerCountAll() int {
	var consumerCount int
	for _, consumers := range es.consumers {
		consumerCount += len(consumers)
	}
	return consumerCount
}
func (es *eventSource) Channels() []string {
	channels := make([]string, 0)
	for channel, _ := range es.consumers {
		channels = append(channels, channel)
	}
	sort.Strings(channels)
	return channels
}
func (es *eventSource) Close(channel string) {
	es.closeChannel <- channel
}
func (es *eventSource) CloseAll() {
	es.closeChannel <- globalChannel
}
func (es *eventSource) Run() {
	runtime.GOMAXPROCS(runtime.NumCPU())
	router := es.Router()
	log.Printf("[I] Starting EventSource service on %s:%d\n", es.settings.GetHost(), es.settings.GetPort())
	log.Fatal("[E]", http.ListenAndServe(fmt.Sprintf("%s:%d", es.settings.GetHost(), es.settings.GetPort()), router))
}
func (es *eventSource) Stop() {
	es.stopApplication <- true
}
func (es *eventSource) subscribeHandler(rw http.ResponseWriter, req *http.Request) {
	params := mux.Vars(req)
	if channel := params["channel"]; len(channel) > 0 {
		if channel == globalChannel {
			log.Printf("[E] Subscribing consumer on %s to global notification channel 'all' rejected\n", req.RemoteAddr)
			http.Error(rw, "Error: Channel 'all' is reserved for global notifications. Please choose another channel name.", http.StatusBadRequest)
			return
		}
		cr, err := newConsumer(rw, req, es, channel)
		if err != nil {
			log.Printf("[E] Subscribing consumer on %s to channel '%s' failed\n", req.RemoteAddr, channel)
			http.Error(rw, fmt.Sprintf("[E] Unable to connect to channel '%s'.", channel), http.StatusInternalServerError)
			return
		}
		es.addConsumer <- cr
	}
}
func (es *eventSource) publishHandler(rw http.ResponseWriter, req *http.Request) {
	if !es.Authenticated(req) {
		log.Printf("[E] Authentication of %s failed. Publishing to channel rejected\n", req.RemoteAddr)
		http.Error(rw, "Error: Authentication failed. Publishing to channel rejected.", http.StatusForbidden)
		return
	}
	if !validContentType(req.Header.Get("Content-Type")) {
		log.Printf("[E] Invalid Content-Type sent by %s. Expecting application/json\n", req.RemoteAddr)
		http.Error(rw, "Error: Invalid Content-Type. Expecting application/json.", http.StatusBadRequest)
		return
	}
	params := mux.Vars(req)
	if channel := params["channel"]; len(channel) > 0 {
		es.SendMessage(req.Body, channel)
		defer req.Body.Close()
	}
	rw.WriteHeader(http.StatusCreated)
}
func (es *eventSource) closeHandler(rw http.ResponseWriter, req *http.Request) {
	if !es.Authenticated(req) {
		log.Printf("[E] Authentication of %s failed. Closing of channel rejected\n", req.RemoteAddr)
		http.Error(rw, "Error: Authentication failed. Closing of channel rejected.", http.StatusForbidden)
		return
	}
	params := mux.Vars(req)
	if channel := params["channel"]; len(channel) > 0 {
		es.Close(channel)
	}
	rw.WriteHeader(http.StatusOK)
}
func (es *eventSource) informationHandler(rw http.ResponseWriter, req *http.Request) {
	if !es.Authenticated(req) {
		log.Printf("[E] Authentication of %s failed. Gettings stats for channel rejected\n", req.RemoteAddr)
		http.Error(rw, "Error: Authentication failed. Gettings stats for channel rejected.", http.StatusForbidden)
		return
	}
	params := mux.Vars(req)
	if channel := params["channel"]; len(channel) > 0 {
		if channel == globalChannel {
			rw.Header().Add("X-Consumer-Count", fmt.Sprint(es.ConsumerCountAll()))
			rw.Header().Add("X-Available-Channels", fmt.Sprintf("[%s]", strings.Join(es.Channels(), ",")))
		} else {
			rw.Header().Add("X-Consumer-Count", fmt.Sprint(es.ConsumerCount(channel)))
			rw.Header().Add("X-Channel-Exists", fmt.Sprint(es.ChannelExists(channel)))
		}
	}
	rw.WriteHeader(http.StatusOK)
}
func channelNotFoundHandler(rw http.ResponseWriter, req *http.Request) {
	log.Printf("[E] Consumer %s tries to join invalid channel", req.RemoteAddr)
	http.Error(rw, "Error: Invalid channel name.", http.StatusNotFound)
}
func (es eventSource) Authenticated(req *http.Request) bool {
	authToken := strings.TrimSpace(req.Header.Get("Auth-Token"))
	if len(es.settings.GetAuthToken()) == 0 && len(authToken) == 0 {
		return true
	}
	return len(es.settings.GetAuthToken()) > 0 && authToken == es.settings.GetAuthToken()
}