<commit_before>// Copyright (c) 2015 Pagoda Box Inc
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v.
// 2.0. If a copy of the MPL was not distributed with this file, You can obtain one
// at http://mozilla.org/MPL/2.0/.
//

package mist

import (
	"fmt"
	"sync"

	"github.com/pagodabox/golang-hatchet"
)

//
const (
	DefaultPort = "1445"
	Version     = "0.0.1"
)

//
type (

	//
	Mist struct {
		sync.Mutex

		log           hatchet.Logger //
		port          string
		Subscriptions map[string]map[chan Message]string //
	}

	//
	Subscription struct {
		Tags []string     `json:"tags"`
		Sub  chan Message `json:"channel"`
	}

	//
	Message struct {
		Tags []string `json:"tags"`
		Data string   `json:"data,string"`
	}
)

//
func New(port string, logger hatchet.Logger) *Mist {

	//
	if logger == nil {
		logger = hatchet.DevNullLogger{}
	}

	mist := &Mist{
		log:           logger,
		port:          port,
		Subscriptions: make(map[string]map[chan Message]string),
	}

	mist.start()

	return mist
}

// Publish takes a list of tags and iterates through Mist's list of subscriptions,
// looking for matching subscriptions to publish messages too. It ensures that the
// list of recipients is a unique set, so as not to publish the same message more
// than once over a channel
func (m *Mist) Publish(tags []string, data string) {

	m.log.Info("PUBLISH!! %v, %v", tags, data)

	// create a message
	msg := Message{Tags: tags, Data: data}

	m.log.Info("MSG: %#v\n", msg)

	// a unique list of recipients (may contain duplicate channels from multiple
	// subscriptions)
	recipients := make(map[chan Message]int)

	// iterate through each provided tag looking for subscriptions to publish to
	for _, tag := range tags {

		// keep track of how many times a subscription is requested
		used := 0

		// iterate through any matching subscriptions and add all of that subscriptions
		// channels to the list of recipients
		if sub, ok := m.Subscriptions[tag]; ok {

			for ch, _ := range sub {

				// ensure that we keep the list of recipients unique, by checking each
				// match against a temporary map of found channels.
				if _, ok := recipients[ch]; !ok {
					used++

					m.log.Info("PUBLISH!!! %#v\n", msg)

					//
					m.log.Debug("[MIST] Publishing: %+v\n", msg)
					go func(ch chan Message) { ch <- msg }(ch)

					// update our list of found channels, with a value of how many times
					// that channel has been subscribed to
					recipients[ch] = used
				}
			}
		}
	}
}

// Subscribe
func (m *Mist) Subscribe(sub Subscription) {
	m.Lock()

	//
	m.log.Debug("[MIST] Subscribe: %+v\n", sub.Tags)

	// iterate over each subscription, adding it to our list of subscriptions (if
	// not already found), and then adding the channel into the subscription's list
	// of subscribers.
	for _, tag := range sub.Tags {

		// if we don't find a subscription, make one (type []chan Message), and add
		// it to our list of subscriptions
		if _, ok := m.Subscriptions[tag]; !ok {
			m.Subscriptions[tag] = make(map[chan Message]string)
			m.log.Debug("[MIST] Created new subscription '%+v'\n", tag)
		}

		// add the channel to each subscription...
		m.Subscriptions[tag][sub.Sub] = ""
		m.log.Debug("[MIST] Subscribed '%+v' to '%+v'\n", sub.Sub, tag)
	}

	m.Unlock()
}

// Unsubscribe
func (m *Mist) Unsubscribe(sub Subscription) {
	m.Lock()

	//
	m.log.Debug("[MIST] Unsubscribing '%+v' from '%+v'\n", sub.Sub, sub.Tags)

	//
	for _, tag := range sub.Tags {

		//
		if s, ok := m.Subscriptions[tag]; ok {
			delete(s, sub.Sub)
			m.log.Debug("[MIST] Unsubscribed '%+v' from '%+v'\n", sub.Sub, s)
		}

		//
		if len(m.Subscriptions[tag]) <= 0 {
			delete(m.Subscriptions, tag)
			m.log.Debug("[MIST] Removed empty subscription '%+v'\n", tag)
		}
	}

	m.Unlock()
}

// List
func (m *Mist) List() {
	fmt.Println(m.Subscriptions)
}<commit_msg>make a minor refactor to the way the client handles data coming in. This should allow tag filtering to work  properly<commit_after>// Copyright (c) 2015 Pagoda Box Inc
//
// This Source Code Form is subject to the terms of the Mozilla Public License, v.
// 2.0. If a copy of the MPL was not distributed with this file, You can obtain one
// at http://mozilla.org/MPL/2.0/.
//

package mist

import (
	"fmt"
	"sync"

	"github.com/pagodabox/golang-hatchet"
)

//
const (
	DefaultPort = "1445"
	Version     = "0.0.1"
)

//
type (

	//
	Mist struct {
		sync.Mutex

		log           hatchet.Logger //
		port          string
		Subscriptions []Subscription //
	}

	//
	Subscription struct {
		Tags []string     `json:"tags"`
		Sub  chan Message `json:"channel"`
	}

	//
	Message struct {
		Tags []string `json:"tags"`
		Data string   `json:"data,string"`
	}
)

//
func New(port string, logger hatchet.Logger) *Mist {

	//
	if logger == nil {
		logger = hatchet.DevNullLogger{}
	}

	mist := &Mist{
		log:           logger,
		port:          port,
		Subscriptions: []Subscription{},
	}

	mist.start()

	return mist
}

// Publish takes a list of tags and iterates through Mist's list of subscriptions,
// looking for matching subscriptions to publish messages too. It ensures that the
// list of recipients is a unique set, so as not to publish the same message more
// than once over a channel
func (m *Mist) Publish(tags []string, data string) {

	m.log.Info("PUBLISH!! %v, %v", tags, data)

	// create a message
	msg := Message{Tags: tags, Data: data}

	m.log.Info("MSG: %#v\n", msg)

	for _, subscription := range m.Subscriptions {
		if contains(tags, subscription.Tags) {
			go func(ch chan Message, msg Message) { ch <- msg }(subscription.Sub, msg)
		}
	}
}

// Subscribe
func (m *Mist) Subscribe(sub Subscription) {
	m.Lock()
	m.Subscriptions = append(m.Subscriptions, sub)
	m.Unlock()
}

// Unsubscribe
func (m *Mist) Unsubscribe(sub Subscription) {
	m.Lock()

	newSubscriptions := []Subscription{}
	for _, subscription := range m.Subscriptions {
		if !sameSub(subscription, sub) {
			newSubscriptions = append(newSubscriptions, sub)
		}
	}
	m.Subscriptions = newSubscriptions

	m.Unlock()
}

// List
func (m *Mist) List() {
	fmt.Println(m.Subscriptions)
}

func contains(full, subset []string) bool {
	fullMap := map[string]interface{}{}
	for _, str := range full {
		fullMap[str] = ""
	}
	for _, str := range subset {
		if fullMap[str] == nil {
			return false
		}
	}

	return true
}

func sameSub(x, y Subscription) bool {
	if len(x.Tags) != len(y.Tags) || x.Sub != y.Sub {
		return false
	}
	for i := 0; i < len(x.Tags); i++ {
		if x.Tags[i] != y.Tags[i] {
			return false
		}
	}

	return true
}
{'fraction_non_alphanumeric': 0.14052984165651644, 'fraction_numerical': 0.0050243605359317906, 'mean_word_length': 2.9217910447761195, 'pattern_counts': {'":': 0, '<': 7, '<?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': '1471919', 'n_tokens_mistral': 2441, 'n_tokens_neox': 2226, 'n_words': 732}