<commit_before>package agent

import (
	"bytes"
	"crypto/tls"
	"errors"
	"fmt"
	"net/url"
	"strings"
	"sync"
	"time"

	mqtt "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
	"github.com/juju/loggo"
)

var AlreadyConfigured = errors.New("Already configured")
var AlreadyUnConfigured = errors.New("Already unconfigured")

//
// Acts as a bridge between local and cloud brokers, this includes reconnecting
// and emitting status changes.
//
// Once configured and started this will attempt to connect to connect to local
// and cloud brokers, if something dies it will reconnect based on the configured
// reconnect backoff.
//
type Bridge struct {
	conf   *Config
	local  *mqtt.MqttClient
	remote *mqtt.MqttClient
	log    loggo.Logger

	localTopics []replaceTopic
	cloudTopics []replaceTopic

	cloudUrl *url.URL
	token    string

	timer       *time.Timer
	reconnectCh chan bool
	shutdownCh  chan bool

	Configured bool
	Connected  bool
	Counter    int64

	IngressCounter int64
	EgressCounter  int64

	IngressBytes int64
	EgressBytes  int64

	LastError error

	bridgeLock sync.Mutex
}

type replaceTopic struct {
	on      string
	replace string
	with    string
}

func (r *replaceTopic) updated(originalTopic string) string {
	return strings.Replace(originalTopic, r.replace, r.with, 1)
}

var localTopics = []replaceTopic{
	// location related topics (TODO: move to cloud userspace RPC)
	{on: "$location/calibration", replace: "$location", with: "$cloud/location"},
	{on: "$location/delete", replace: "$location", with: "$cloud/location"},
	{on: "$device/+/+/rssi", replace: "$device", with: "$cloud/device"},

	// module health statistics
	//{on: "$node/+/module/status", replace: "$node", with: "$cloud/node"},

	// cloud userspace RPC requests
	{on: "$ninja/services/rpc/+/+", replace: "$ninja", with: "$cloud/ninja"},
	{on: "$ninja/services/+", replace: "$ninja", with: "$cloud/ninja"},

	// temporary alternate topic to distinguish remote device replies from local-destined ones
	// used by the phone app for remote actuations
	// the alternate remote_ topic is to prevent a loopback with the below rule in the other direction
	// TODO: use a tag like $mesh-source to prevent loops (never re-proxy msgs with your source)
	{on: "$device/+/channel/+/reply", replace: "$device", with: "$cloud/remote_device"},

	// push up all local RPC methods in case the cloud is responding,
	// this is currently used for the push notification channel
	{on: "$device/+/channel/+", replace: "$device", with: "$cloud/device"},

	// push up state changes to the cloud
	{on: "$device/+/channel/+/event/state", replace: "$device", with: "$cloud/device"},
}

var cloudTopics = []replaceTopic{
	// location related topics
	{on: "$cloud/location/calibration/progress", replace: "$cloud/location", with: "$location"},
	{on: "$cloud/device/+/+/location", replace: "$cloud/device", with: "$device"},

	// cloud userspace RPC replies
	{on: "$cloud/ninja/services/rpc/+/+/reply", replace: "$cloud/ninja", with: "$ninja"},
	{on: "$cloud/ninja/services/+/reply", replace: "$cloud/ninja", with: "$ninja"},

	// see comment for $device/+/channel/+/reply above
	{on: "$cloud/remote_device/+/channel/+", replace: "$cloud/remote_device", with: "$device"},

	// allow cloud to announce devices and channels (used for phone on 3G and notification subscription channel)
	{on: "$cloud/device/+/event/announce", replace: "$cloud/device", with: "$device"},
	{on: "$cloud/device/+/channel/+/event/announce", replace: "$cloud/device", with: "$device"},

	// retrieve RPC responses from the cloud,
	// this is currently used for the push notification channel
	{on: "$cloud/device/+/channel/+/reply", replace: "$cloud/device", with: "$device"},
}

func createBridge(conf *Config) *Bridge {
	return &Bridge{conf: conf, localTopics: localTopics, cloudTopics: cloudTopics, log: loggo.GetLogger("bridge")}
}

func (b *Bridge) start(cloudUrl string, token string) (err error) {

	if b.Configured {
		b.log.Warningf("Already configured.")
		return AlreadyConfigured
	}<commit_msg>Revert "Add an additional routing."<commit_after>package agent

import (
	"bytes"
	"crypto/tls"
	"errors"
	"fmt"
	"net/url"
	"strings"
	"sync"
	"time"

	mqtt "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
	"github.com/juju/loggo"
)

var AlreadyConfigured = errors.New("Already configured")
var AlreadyUnConfigured = errors.New("Already unconfigured")

//
// Acts as a bridge between local and cloud brokers, this includes reconnecting
// and emitting status changes.
//
// Once configured and started this will attempt to connect to connect to local
// and cloud brokers, if something dies it will reconnect based on the configured
// reconnect backoff.
//
type Bridge struct {
	conf   *Config
	local  *mqtt.MqttClient
	remote *mqtt.MqttClient
	log    loggo.Logger

	localTopics []replaceTopic
	cloudTopics []replaceTopic

	cloudUrl *url.URL
	token    string

	timer       *time.Timer
	reconnectCh chan bool
	shutdownCh  chan bool

	Configured bool
	Connected  bool
	Counter    int64

	IngressCounter int64
	EgressCounter  int64

	IngressBytes int64
	EgressBytes  int64

	LastError error

	bridgeLock sync.Mutex
}

type replaceTopic struct {
	on      string
	replace string
	with    string
}

func (r *replaceTopic) updated(originalTopic string) string {
	return strings.Replace(originalTopic, r.replace, r.with, 1)
}

var localTopics = []replaceTopic{
	// location related topics (TODO: move to cloud userspace RPC)
	{on: "$location/calibration", replace: "$location", with: "$cloud/location"},
	{on: "$location/delete", replace: "$location", with: "$cloud/location"},
	{on: "$device/+/+/rssi", replace: "$device", with: "$cloud/device"},

	// module health statistics
	//{on: "$node/+/module/status", replace: "$node", with: "$cloud/node"},

	// cloud userspace RPC requests
	{on: "$ninja/services/rpc/+/+", replace: "$ninja", with: "$cloud/ninja"},
	{on: "$ninja/services/+", replace: "$ninja", with: "$cloud/ninja"},

	// temporary alternate topic to distinguish remote device replies from local-destined ones
	// used by the phone app for remote actuations
	// the alternate remote_ topic is to prevent a loopback with the below rule in the other direction
	// TODO: use a tag like $mesh-source to prevent loops (never re-proxy msgs with your source)
	{on: "$device/+/channel/+/reply", replace: "$device", with: "$cloud/remote_device"},

	// push up all local RPC methods in case the cloud is responding,
	// this is currently used for the push notification channel
	{on: "$device/+/channel/+", replace: "$device", with: "$cloud/device"},

	// push up state changes to the cloud
	{on: "$device/+/channel/+/event/state", replace: "$device", with: "$cloud/device"},
}

var cloudTopics = []replaceTopic{
	// location related topics
	{on: "$cloud/location/calibration/progress", replace: "$cloud/location", with: "$location"},
	{on: "$cloud/device/+/+/location", replace: "$cloud/device", with: "$device"},

	// cloud userspace RPC replies
	{on: "$cloud/ninja/services/rpc/+/+/reply", replace: "$cloud/ninja", with: "$ninja"},

	// see comment for $device/+/channel/+/reply above
	{on: "$cloud/remote_device/+/channel/+", replace: "$cloud/remote_device", with: "$device"},

	// allow cloud to announce devices and channels (used for phone on 3G and notification subscription channel)
	{on: "$cloud/device/+/event/announce", replace: "$cloud/device", with: "$device"},
	{on: "$cloud/device/+/channel/+/event/announce", replace: "$cloud/device", with: "$device"},

	// retrieve RPC responses from the cloud,
	// this is currently used for the push notification channel
	{on: "$cloud/device/+/channel/+/reply", replace: "$cloud/device", with: "$device"},
}

func createBridge(conf *Config) *Bridge {
	return &Bridge{conf: conf, localTopics: localTopics, cloudTopics: cloudTopics, log: loggo.GetLogger("bridge")}
}

func (b *Bridge) start(cloudUrl string, token string) (err error) {

	if b.Configured {
		b.log.Warningf("Already configured.")
		return AlreadyConfigured
	}
{'fraction_non_alphanumeric': 0.14764098095356654, 'fraction_numerical': 0.002987675837171667, 'mean_word_length': 4.391946308724832, 'pattern_counts': {'":': 0, '<': 3, '<?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': '81305', 'n_tokens_mistral': 2628, 'n_tokens_neox': 2459, 'n_words': 808}