func (p *Proxy) handleMessages() {

	defer func() {
		lumber.Trace("Got p.done, closing check and pipe")
		close(p.check)
		close(p.Pipe) // don't close pipe (response/pong messages need it), but leaving it unclosed leaves ram bloat on server even after client disconnects
	}()

	//
	for {
		select {

		// we need to ensure that this subscription actually has these tags before
		// sending anything to it; not doing this will cause everything to come
		// across the channel
		case msg := <-p.check:
			lumber.Trace("Got p.check")
			p.RLock()
			match := p.subscriptions.Match(msg.Tags)
			p.RUnlock()

			// if there is a subscription for the tags publish the message
			if match {
				lumber.Trace("Sending msg on pipe")
				p.Pipe <- msg
			}

		//
		case <-p.done:
			return
		}
	}
}

// SetTask adds or updates a kapacitor task
func SetTask(task Task) error {
	var Type client.TaskType
	var Status client.TaskStatus
	DBRPs := make([]client.DBRP, 1)

	// convert type
	switch strings.ToUpper(task.Type) {
	case "BATCH":
		Type = client.BatchTask
	case "STREAM":
		Type = client.StreamTask
	default:
		return fmt.Errorf("Bad task type - '%v'", task.Type)
	}

	DBRPs[0].Database = task.Database
	DBRPs[0].RetentionPolicy = task.RetentionPolicy

	// convert status
	switch strings.ToUpper(task.Status) {
	case "DISABLED":
		Status = client.Disabled
	case "ENABLED":
		Status = client.Enabled
	case "":
		// default to disabled
		Status = client.Disabled
	default:
		return fmt.Errorf("Bad task status - '%v'", task.Status)
	}

	var err error
	l := cli.TaskLink(task.Id)
	t, _ := cli.Task(l, nil)
	if t.ID == "" {
		_, err = cli.CreateTask(client.CreateTaskOptions{
			ID:         task.Id,
			Type:       Type,
			DBRPs:      DBRPs,
			TICKscript: task.Script,
			Status:     Status,
		})
		lumber.Trace("Task Created")
	} else {
		_, err = cli.UpdateTask(
			l,
			client.UpdateTaskOptions{
				Type:       Type,
				DBRPs:      DBRPs,
				TICKscript: task.Script,
			},
		)
		lumber.Trace("Task Updated")
	}
	if err != nil {
		return fmt.Errorf("Failed to create task - %v", err)
	}

	return nil
}

// runLoop handles communication from the server
func (relay *Relay) runLoop(reader *bufio.Reader) {
	for {
		// when implementing relay, set `lumber.Level(lumber.LvlInt("TRACE"))` in client to view logs
		line, err := reader.ReadString('\n')
		if err != nil {
			lumber.Error("[PULSE :: RELAY] Disconnected from host %v!", relay.hostAddr)
			// retry indefinitely
			for {
				if reader, err = relay.establishConnection(); err == nil {
					lumber.Info("[PULSE :: RELAY] Reconnected to host %v!", relay.hostAddr)
					break
				}
				lumber.Debug("[PULSE :: RELAY] Reconnecting to host %v...  Fail!", relay.hostAddr)
				<-time.After(5 * time.Second)
			}
			// we won't have anything in 'line' so continue
			continue
		}

		line = strings.TrimSuffix(line, "\n")
		split := strings.SplitN(line, " ", 2)

		cmd := split[0]
		switch cmd {
		case "ok":
			lumber.Trace("[PULSE :: RELAY] OK: %v", split)
			// just an ack
		case "get":
			lumber.Trace("[PULSE :: RELAY] GET: %v", split)
			if len(split) != 2 {
				continue
			}
			stats := strings.Split(split[1], ",")
			results := make([]string, 0)
			for _, stat := range stats {
				tagCollector, ok := relay.collectors[stat]
				if !ok {
					continue
				}
				for name, value := range tagCollector.collector.Collect() {
					formatted := strconv.FormatFloat(value, 'f', 4, 64)
					if name == "" {
						name = stat
					}
					results = append(results, fmt.Sprintf("%s-%s:%s", stat, name, formatted))
				}
			}
			response := fmt.Sprintf("got %s\n", strings.Join(results, ","))
			relay.conn.Write([]byte(response))
		default:
			lumber.Trace("[PULSE :: RELAY] BAD: %v", split)
			relay.conn.Write([]byte("unknown command\n"))
		}
	}
}

// PollAll polls all clients for registered collectors(stats to be collected)
func PollAll() {
	lumber.Trace("[PULSE :: SERVER] PollAll...")
	for _, client := range clients {
		command := "get " + strings.Join(client.collectorList(), ",") + "\n"
		go client.conn.Write([]byte(command))
	}
}

// subscribe adds a proxy to the list of mist subscribers; we need this so that
// we can lock this process incase multiple proxies are subscribing at the same
// time
func subscribe(p *Proxy) {
	lumber.Trace("Adding proxy to subscribers...")

	mutex.Lock()
	subscribers[p.id] = p
	mutex.Unlock()
}

// unsubscribe removes a proxy from the list of mist subscribers; we need this
// so that we can lock this process incase multiple proxies are unsubscribing at
// the same time
func unsubscribe(pid uint32) {
	lumber.Trace("Removing proxy from subscribers...")

	mutex.Lock()
	delete(subscribers, pid)
	mutex.Unlock()
}

// List returns a list of all current subscriptions
func (p *Proxy) List() (data [][]string) {
	lumber.Trace("Proxy listing subscriptions...")
	p.RLock()
	data = p.subscriptions.ToSlice()
	p.RUnlock()

	return
}

func (c *Client) Send() {
	for msg := range c.messages {
		log.Trace("Sending message: %s", msg)
		if err := websocket.Message.Send(c.ws, msg); err != nil {
			log.Error("Error sending message: %s", err)
		}
	}
}

func (c *Client) Publish(topicURI string, event interface{}, opts ...interface{}) error {
	log.Trace("sending publish)")
	msg, err := CreatePublish(topicURI, event, opts...)
	if err != nil {
		return fmt.Errorf("turnpike: %s", err)
	}
	c.messages <- string(msg)
	return nil
}

func (c *Client) Unsubscribe(topicURI string) error {
	log.Trace("sending unsubscribe")
	msg, err := CreateUnsubscribe(topicURI)
	if err != nil {
		return fmt.Errorf("turnpike: %s", err)
	}
	c.messages <- string(msg)
	return nil
}

func (c *Client) Call(callID, procURI string, args ...interface{}) error {
	log.Trace("sending call")
	msg, err := CreateCall(callID, procURI, args...)
	if err != nil {
		return fmt.Errorf("turnpike: %s", err)
	}
	c.messages <- string(msg)
	return nil
}

func Query(sql string) (*client.Response, error) {
	lumber.Trace("[PULSE :: INFLUX] Querying influx: '%v'...", sql)

	c, err := influxClient()
	if err != nil {
		return nil, err
	}
	return c.Query(client.NewQuery(fmt.Sprint(sql), "statistics", "s"))
}

// AddCollector adds a collector to relay
func (relay *Relay) AddCollector(name string, tags []string, collector Collector) error {
	if name == "_connected" || strings.ContainsAny(name, "-:,") {
		lumber.Trace("[PULSE :: RELAY] Reserved name!")
		return ReservedName
	}
	if _, ok := relay.collectors[name]; ok {
		lumber.Trace("[PULSE :: RELAY] Duplicate collector!")
		return DuplicateCollector
	}
	if _, err := relay.conn.Write([]byte(fmt.Sprintf("add %s:%s\n", name, strings.Join(tags, ",")))); err != nil {
		lumber.Trace("[PULSE :: RELAY] Failed to write!")
		return err
	}

	// if successfully added collector, add it to relay's known collectors
	relay.collectors[name] = taggedCollector{collector: collector, tags: tags}
	return nil
}

func (t *Server) handleUnsubscribe(id string, msg UnsubscribeMsg) {
	log.Trace("Handling unsubscribe message")
	t.subLock.Lock()
	topic := CheckCurie(t.prefixes[id], msg.TopicURI)
	if lm, ok := t.subscriptions[topic]; ok {
		lm.Remove(id)
	}
	t.subLock.Unlock()
	log.Debug("Client %s unsubscribed from topic: %s", id, topic)
}

// Poll polls based on tags
func Poll(tags []string) {
	lumber.Trace("[PULSE :: SERVER] Poll...")
	if tags == nil {
		PollAll()
		return
	}
	ids := findIds(tags)
	command := "get " + strings.Join(tags, ",") + "\n"
	sendAll(command, ids)
}

func (t *Server) handlePrefix(id string, msg PrefixMsg) {
	log.Trace("Handling prefix message")
	if _, ok := t.prefixes[id]; !ok {
		t.prefixes[id] = make(PrefixMap)
	}
	if err := t.prefixes[id].RegisterPrefix(msg.Prefix, msg.URI); err != nil {
		log.Error("Error registering prefix: %s", err)
	}
	log.Debug("Client %s registered prefix '%s' for URI: %s", id, msg.Prefix, msg.URI)
}

// publish publishes to all subscribers except the one who issued the publish
func publish(pid uint32, tags []string, data string) error {

	if len(tags) == 0 {
		return fmt.Errorf("Failed to publish. Missing tags")
	}

	// if there are no subscribers, the message goes nowhere
	//
	// this could be more optimized, but it might not be an issue unless thousands
	// of clients are using mist.
	go func() {
		mutex.RLock()
		for _, subscriber := range subscribers {
			select {
			case <-subscriber.done:
				lumber.Trace("Subscriber done")
				// do nothing?

			default:

				// dont send this message to the publisher who just sent it
				if subscriber.id == pid {
					lumber.Trace("Subscriber is publisher, skipping publish")
					continue
				}

				// create message
				msg := Message{Command: "publish", Tags: tags, Data: data}

				// we don't want this operation blocking the range of other subscribers
				// waiting to get messages
				go func(p *Proxy, msg Message) {
					p.check <- msg
					lumber.Trace("Published message")
				}(subscriber, msg)
			}
		}
		mutex.RUnlock()
	}()

	return nil
}

// Close ...
func (p *Proxy) Close() {
	lumber.Trace("Proxy closing...")

	if len(p.subscriptions.ToSlice()) != 0 {
		// remove the local p from mist's list of subscribers
		unsubscribe(p.id)
	}

	// this closes the goroutine that is matching messages to subscriptions
	close(p.done)
}

func (t *Server) handleSubscribe(id string, msg SubscribeMsg) {
	log.Trace("Handling subscribe message")
	t.subLock.Lock()
	topic := CheckCurie(t.prefixes[id], msg.TopicURI)
	if _, ok := t.subscriptions[topic]; !ok {
		t.subscriptions[topic] = make(map[string]bool)
	}
	t.subscriptions[topic].Add(id)
	t.subLock.Unlock()
	log.Debug("Client %s subscribed to topic: %s", id, topic)
}

// Unsubscribe ...
func (p *Proxy) Unsubscribe(tags []string) {
	lumber.Trace("Proxy unsubscribing from '%v'...", tags)

	if len(tags) == 0 {
		return
	}

	// remove tags from subscription
	p.Lock()
	p.subscriptions.Remove(tags)
	p.Unlock()
}