<commit_before>package dispatcher

import (
	"reflect"

	log "github.com/Sirupsen/logrus"
	"github.com/openskybot/skybot-router/common"
)

// ChanQueueLength buffered channel length
const ChanQueueLength = 100

// Object native representation of a UAVPacket, just a map
type Object map[string]interface{}

// Update is the UAVTalk protocol packet, encapsulates a common in the field Data
type Update struct {
	ObjectID   uint32 `json:"objectId"`
	InstanceID uint16 `json:"instanceId"`
	Data       Object `json:"data"`
}

// Request is the packet that requests a common data, is forwarded to the owner of a definition (the one that sent the definition)
type Request struct {
	ObjectID   uint32 `json:"objectId"`
	InstanceID uint16 `json:"instanceId"`
}

// Subscription adds an objectID to the subscriptions of the sending connection
type Subscription struct {
	ObjectID uint32 `json:"objectId"`
}

// Unsubscription removes an objectID from the subscriptions of the sending connection
type Unsubscription struct {
	ObjectID uint32 `json:"objectId"`
}

// Connection : basic interface representing a connection to the dispatcher
type Connection struct {
	definitions   common.Definitions
	subscriptions []uint32
	InChan        chan interface{}
	OutChan       chan interface{}
}

// NewConnection creates a new dispatcher connection
func NewConnection() *Connection {
	connection := new(Connection)
	connection.InChan = make(chan interface{}, ChanQueueLength)
	connection.OutChan = make(chan interface{}, ChanQueueLength)

	return connection
}

// Close closes the connection, possible issues...
func (connection *Connection) Close() {
	close(connection.OutChan)
}

func (connection *Connection) addSubscription(objectID uint32) {
	connection.subscriptions = append(connection.subscriptions, objectID)
}

func (connection *Connection) removeSubscription(objectID uint32) {
	for i, subscription := range connection.subscriptions {
		if subscription == objectID {
			if i < len(connection.subscriptions)-1 {
				copy(connection.subscriptions[i:], connection.subscriptions[i+1:])
			}
			connection.subscriptions = connection.subscriptions[0 : len(connection.subscriptions)-1]
			return
		}
	}
}

// Dispatcher main dispatcher class
type Dispatcher struct {
	connections    []*Connection
	cases          []reflect.SelectCase // cases for the select case of the main loop, the first element il for the connectionChan, the others are for the outChans of the connections
	connectionChan chan *Connection     // connectionChan receives the new connections to add
}

// NewDispatcher creates a dispatcher
func NewDispatcher() *Dispatcher {
	dispatcher := new(Dispatcher)
	dispatcher.connections = make([]*Connection, 0, 100)
	dispatcher.cases = make([]reflect.SelectCase, 0, 100)
	dispatcher.connectionChan = make(chan *Connection, 10) // TODO try unbuffered chan

	// first case is for the connectionChan
	dispatcher.cases = append(dispatcher.cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(dispatcher.connectionChan)})

	return dispatcher
}

// AddConnection adds a connection to the dispatcher
func (dispatcher *Dispatcher) AddConnection(connection *Connection) {
	dispatcher.connectionChan <- connection
}

func (dispatcher *Dispatcher) addConnection(connection *Connection) {
	for _, c := range dispatcher.connections {
		for _, d := range c.definitions {
			connection.InChan <- *d
		}
	}

	dispatcher.connections = append(dispatcher.connections, connection)
	dispatcher.cases = append(dispatcher.cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(connection.OutChan)})
}

func (dispatcher *Dispatcher) removeConnectionAt(index int) {
	// if it is not the last element, move all next elements
	if index < len(dispatcher.connections) {
		copy(dispatcher.connections[index:], dispatcher.connections[index+1:])
		copy(dispatcher.cases[index+1:], dispatcher.cases[index+2:])
	}
	dispatcher.connections[len(dispatcher.connections)-1] = nil
	dispatcher.connections = dispatcher.connections[:len(dispatcher.connections)-1]

	dispatcher.cases = dispatcher.cases[:len(dispatcher.cases)-1]
}

func (dispatcher *Dispatcher) dispatchUpdate(from int, update *Update) {
	for i, connection := range dispatcher.connections {
		if i == from {
			continue
		}
		subscribed := false
		for _, objectID := range connection.subscriptions {
			if objectID == update.ObjectID {
				subscribed = true
			}
		}
		if subscribed == false {
			if _, err := connection.definitions.GetDefinitionForObjectID(update.ObjectID); err != nil {
				continue
			}
		}
		connection.InChan <- *update
	}<commit_msg>only dispatcher update message to definitions owner when update originiates to a non-owner<commit_after>package dispatcher

import (
	"reflect"

	log "github.com/Sirupsen/logrus"
	"github.com/openskybot/skybot-router/common"
)

// ChanQueueLength buffered channel length
const ChanQueueLength = 100

// Object native representation of a UAVPacket, just a map
type Object map[string]interface{}

// Update is the UAVTalk protocol packet, encapsulates a common in the field Data
type Update struct {
	ObjectID   uint32 `json:"objectId"`
	InstanceID uint16 `json:"instanceId"`
	Data       Object `json:"data"`
}

// Request is the packet that requests a common data, is forwarded to the owner of a definition (the one that sent the definition)
type Request struct {
	ObjectID   uint32 `json:"objectId"`
	InstanceID uint16 `json:"instanceId"`
}

// Subscription adds an objectID to the subscriptions of the sending connection
type Subscription struct {
	ObjectID uint32 `json:"objectId"`
}

// Unsubscription removes an objectID from the subscriptions of the sending connection
type Unsubscription struct {
	ObjectID uint32 `json:"objectId"`
}

// Connection : basic interface representing a connection to the dispatcher
type Connection struct {
	definitions   common.Definitions
	subscriptions []uint32
	InChan        chan interface{}
	OutChan       chan interface{}
}

// NewConnection creates a new dispatcher connection
func NewConnection() *Connection {
	connection := new(Connection)
	connection.InChan = make(chan interface{}, ChanQueueLength)
	connection.OutChan = make(chan interface{}, ChanQueueLength)

	return connection
}

// Close closes the connection, possible issues...
func (connection *Connection) Close() {
	close(connection.OutChan)
}

func (connection *Connection) addSubscription(objectID uint32) {
	connection.subscriptions = append(connection.subscriptions, objectID)
}

func (connection *Connection) removeSubscription(objectID uint32) {
	for i, subscription := range connection.subscriptions {
		if subscription == objectID {
			if i < len(connection.subscriptions)-1 {
				copy(connection.subscriptions[i:], connection.subscriptions[i+1:])
			}
			connection.subscriptions = connection.subscriptions[0 : len(connection.subscriptions)-1]
			return
		}
	}
}

// Dispatcher main dispatcher class
type Dispatcher struct {
	connections    []*Connection
	cases          []reflect.SelectCase // cases for the select case of the main loop, the first element il for the connectionChan, the others are for the outChans of the connections
	connectionChan chan *Connection     // connectionChan receives the new connections to add
}

// NewDispatcher creates a dispatcher
func NewDispatcher() *Dispatcher {
	dispatcher := new(Dispatcher)
	dispatcher.connections = make([]*Connection, 0, 100)
	dispatcher.cases = make([]reflect.SelectCase, 0, 100)
	dispatcher.connectionChan = make(chan *Connection, 10) // TODO try unbuffered chan

	// first case is for the connectionChan
	dispatcher.cases = append(dispatcher.cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(dispatcher.connectionChan)})

	return dispatcher
}

// AddConnection adds a connection to the dispatcher
func (dispatcher *Dispatcher) AddConnection(connection *Connection) {
	dispatcher.connectionChan <- connection
}

func (dispatcher *Dispatcher) addConnection(connection *Connection) {
	for _, c := range dispatcher.connections {
		for _, d := range c.definitions {
			connection.InChan <- *d
		}
	}

	dispatcher.connections = append(dispatcher.connections, connection)
	dispatcher.cases = append(dispatcher.cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(connection.OutChan)})
}

func (dispatcher *Dispatcher) removeConnectionAt(index int) {
	// if it is not the last element, move all next elements
	if index < len(dispatcher.connections) {
		copy(dispatcher.connections[index:], dispatcher.connections[index+1:])
		copy(dispatcher.cases[index+1:], dispatcher.cases[index+2:])
	}
	dispatcher.connections[len(dispatcher.connections)-1] = nil
	dispatcher.connections = dispatcher.connections[:len(dispatcher.connections)-1]

	dispatcher.cases = dispatcher.cases[:len(dispatcher.cases)-1]
}

func (dispatcher *Dispatcher) dispatchUpdate(from int, update *Update) {
	fromConnection := dispatcher.connections[from]
	fromOwner := false
	if _, err := fromConnection.definitions.GetDefinitionForObjectID(update.ObjectID); err == nil {
		fromOwner = true
	}
mainLoop:
	for i, connection := range dispatcher.connections {
		if i == from {
			continue
		}

		if fromOwner == true {
			for _, objectID := range connection.subscriptions {
				if objectID == update.ObjectID {
					connection.InChan <- *update
					continue mainLoop
				}
			}
		} else {
			if _, err := connection.definitions.GetDefinitionForObjectID(update.ObjectID); err == nil {
				connection.InChan <- *update
			}
		}
	}
{'fraction_non_alphanumeric': 0.09743805123897523, 'fraction_numerical': 0.008609827803443932, 'mean_word_length': 4.815018315018315, 'pattern_counts': {'":': 0, '<': 14, '<?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': '2098135', 'n_tokens_mistral': 2877, 'n_tokens_neox': 2604, 'n_words': 806}