<commit_before>package agent

import (
	"crypto/tls"
	"fmt"
	"log"
	"strings"

	mqtt "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
)

type Agent struct {
	conf   *Config
	local  *mqtt.MqttClient
	remote *mqtt.MqttClient

	localTopics []replaceTopic
	cloudTopics []replaceTopic
}

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{
	{on: "$location/calibration", replace: "$location", with: "$cloud/location"},
	{on: "$device/+/+/rssi", replace: "$device", with: "$cloud/device"},
}

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

func createAgent(conf *Config) *Agent {
	return &Agent{conf: conf, localTopics: localTopics, cloudTopics: cloudTopics}
}

func (a *Agent) start() error {

	a.local = a.buildClient(a.conf.LocalUrl, "")
	a.remote = a.buildClient(a.conf.CloudUrl, a.conf.Token)

	// this is a very basic pipe between the two
	// need to do some work on reconnects and receipt handling
	a.subscribe(a.local, a.remote, a.localTopics, "local")
	a.subscribe(a.remote, a.local, a.cloudTopics, "cloud")

	return nil
}

func (a *Agent) buildClient(server string, token string) *mqtt.MqttClient {

	opts := mqtt.NewClientOptions().SetBroker(server).SetTraceLevel(mqtt.Off).SetTlsConfig(&tls.Config{InsecureSkipVerify: true})

	if token != "" {
		opts.SetUsername(token)
	}

	// pretty much log the reason and quit
	opts.SetOnConnectionLost(a.onConnectionLoss)

	client := mqtt.NewClient(opts)
	_, err := client.Start()
	if err != nil {
		log.Fatalf("error starting connection: %s", err)
	} else {
		fmt.Printf("Connected as %s\n", server)
	}
	return client
}

func (a *Agent) subscribe(src *mqtt.MqttClient, dst *mqtt.MqttClient, topics []replaceTopic, tag string) {
	for _, topic := range topics {

		topicFilter, _ := mqtt.NewTopicFilter(topic.on, 0)
		log.Printf("[%s] subscribed to %s", tag, topic.on)

		if _, err := src.StartSubscription(func(src *mqtt.MqttClient, msg mqtt.Message) {
			if a.conf.IsDebug() {
				log.Printf("[%s] topic: %s updated: %s len: %d", tag, msg.Topic(), topic.updated(msg.Topic()), len(msg.Payload()))
			}
			dst.PublishMessage(topic.updated(msg.Topic()), mqtt.NewMessage(msg.Payload()))
		}, topicFilter); err != nil {
			log.Fatalf("error starting subscription: %s", err)
		}
	}
}

func (a *Agent) onConnectionLoss(client *mqtt.MqttClient, reason error) {
	log.Fatalf("Connection failed %s", reason)
}
<commit_msg>Update calibration topic to progress instead of complete<commit_after>package agent

import (
	"crypto/tls"
	"fmt"
	"log"
	"strings"

	mqtt "git.eclipse.org/gitroot/paho/org.eclipse.paho.mqtt.golang.git"
)

type Agent struct {
	conf   *Config
	local  *mqtt.MqttClient
	remote *mqtt.MqttClient

	localTopics []replaceTopic
	cloudTopics []replaceTopic
}

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{
	{on: "$location/calibration", replace: "$location", with: "$cloud/location"},
	{on: "$device/+/+/rssi", replace: "$device", with: "$cloud/device"},
}

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

func createAgent(conf *Config) *Agent {
	return &Agent{conf: conf, localTopics: localTopics, cloudTopics: cloudTopics}
}

func (a *Agent) start() error {

	a.local = a.buildClient(a.conf.LocalUrl, "")
	a.remote = a.buildClient(a.conf.CloudUrl, a.conf.Token)

	// this is a very basic pipe between the two
	// need to do some work on reconnects and receipt handling
	a.subscribe(a.local, a.remote, a.localTopics, "local")
	a.subscribe(a.remote, a.local, a.cloudTopics, "cloud")

	return nil
}

func (a *Agent) buildClient(server string, token string) *mqtt.MqttClient {

	opts := mqtt.NewClientOptions().SetBroker(server).SetTraceLevel(mqtt.Off).SetTlsConfig(&tls.Config{InsecureSkipVerify: true})

	if token != "" {
		opts.SetUsername(token)
	}

	// pretty much log the reason and quit
	opts.SetOnConnectionLost(a.onConnectionLoss)

	client := mqtt.NewClient(opts)
	_, err := client.Start()
	if err != nil {
		log.Fatalf("error starting connection: %s", err)
	} else {
		fmt.Printf("Connected as %s\n", server)
	}
	return client
}

func (a *Agent) subscribe(src *mqtt.MqttClient, dst *mqtt.MqttClient, topics []replaceTopic, tag string) {
	for _, topic := range topics {

		topicFilter, _ := mqtt.NewTopicFilter(topic.on, 0)
		log.Printf("[%s] subscribed to %s", tag, topic.on)

		if _, err := src.StartSubscription(func(src *mqtt.MqttClient, msg mqtt.Message) {
			if a.conf.IsDebug() {
				log.Printf("[%s] topic: %s updated: %s len: %d", tag, msg.Topic(), topic.updated(msg.Topic()), len(msg.Payload()))
			}
			dst.PublishMessage(topic.updated(msg.Topic()), mqtt.NewMessage(msg.Payload()))
		}, topicFilter); err != nil {
			log.Fatalf("error starting subscription: %s", err)
		}
	}
}

func (a *Agent) onConnectionLoss(client *mqtt.MqttClient, reason error) {
	log.Fatalf("Connection failed %s", reason)
}

{'fraction_non_alphanumeric': 0.15897903372835004, 'fraction_numerical': 0.0007292616226071103, 'mean_word_length': 4.6440329218107, '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': '732585', 'n_tokens_mistral': 1999, 'n_tokens_neox': 1933, 'n_words': 461}