<commit_before>// Copyright © 2016 The Things Network
// Use of this source code is governed by the MIT license that can be found in the LICENSE file.

package collection

import (
	"fmt"
	"time"

	"github.com/TheThingsNetwork/ttn/core"
	"github.com/TheThingsNetwork/ttn/core/types"
	"github.com/TheThingsNetwork/ttn/mqtt"
	"github.com/apex/log"
)

// AppCollector represents a collector for application data
type AppCollector interface {
	Start() error
	Stop()
}

type appCollector struct {
	ctx       log.Interface
	eui       types.AppEUI
	functions *Functions
	broker    string
	client    mqtt.Client
	storage   DataStorage
}

// NewMqttAppCollector instantiates a new AppCollector instance using MQTT
func NewMqttAppCollector(ctx log.Interface, broker string, eui types.AppEUI, key string, functions *Functions, storage DataStorage) AppCollector {
	return &appCollector{
		ctx:       ctx,
		eui:       eui,
		functions: functions,
		broker:    broker,
		client:    mqtt.NewClient(ctx, "collector", eui.String(), key, fmt.Sprintf("tcp://%s", broker)),
		storage:   storage,
	}
}

func (c *appCollector) Start() error {
	err := c.client.Connect()
	if err != nil {
		c.ctx.WithError(err).Error("Connect failed")
		return err
	}
	if token := c.client.SubscribeAppUplink(c.eui, c.handleUplink); token.Wait() && token.Error() != nil {
		c.ctx.WithError(token.Error()).Error("Failed to subscribe")
		return token.Error()
	}
	c.ctx.WithField("broker", c.broker).Info("Subscribed to app uplink packets")
	return nil
}

func (c *appCollector) Stop() {
	c.client.Disconnect()
}

func (c *appCollector) handleUplink(client mqtt.Client, appEUI types.AppEUI, devEUI types.DevEUI, req core.DataUpAppReq) {
	ctx := c.ctx.WithField("devEUI", devEUI)

	fields, valid, err := c.functions.Process(req.Payload)
	if err != nil {
		ctx.WithError(err).Error("Failed to process payload")
		return
	}
	if !valid {
		ctx.Warn("The payload is not valid")
		return
	}

	t, err := time.Parse(time.RFC3339, req.Metadata[0].ServerTime)
	if err != nil {
		ctx.WithError(err).Warnf("Invalid time: %v", req.Metadata[0].ServerTime)
		return
	}

	err = c.storage.Save(appEUI, devEUI, t, fields)
	if err != nil {
		ctx.WithError(err).Error("Failed to save data")
	}
	ctx.Debug("Saved uplink packet in store")
}<commit_msg>Return after error reporting<commit_after>// Copyright © 2016 The Things Network
// Use of this source code is governed by the MIT license that can be found in the LICENSE file.

package collection

import (
	"fmt"
	"time"

	"github.com/TheThingsNetwork/ttn/core"
	"github.com/TheThingsNetwork/ttn/core/types"
	"github.com/TheThingsNetwork/ttn/mqtt"
	"github.com/apex/log"
)

// AppCollector represents a collector for application data
type AppCollector interface {
	Start() error
	Stop()
}

type appCollector struct {
	ctx       log.Interface
	eui       types.AppEUI
	functions *Functions
	broker    string
	client    mqtt.Client
	storage   DataStorage
}

// NewMqttAppCollector instantiates a new AppCollector instance using MQTT
func NewMqttAppCollector(ctx log.Interface, broker string, eui types.AppEUI, key string, functions *Functions, storage DataStorage) AppCollector {
	return &appCollector{
		ctx:       ctx,
		eui:       eui,
		functions: functions,
		broker:    broker,
		client:    mqtt.NewClient(ctx, "collector", eui.String(), key, fmt.Sprintf("tcp://%s", broker)),
		storage:   storage,
	}
}

func (c *appCollector) Start() error {
	err := c.client.Connect()
	if err != nil {
		c.ctx.WithError(err).Error("Connect failed")
		return err
	}
	if token := c.client.SubscribeAppUplink(c.eui, c.handleUplink); token.Wait() && token.Error() != nil {
		c.ctx.WithError(token.Error()).Error("Failed to subscribe")
		return token.Error()
	}
	c.ctx.WithField("broker", c.broker).Info("Subscribed to app uplink packets")
	return nil
}

func (c *appCollector) Stop() {
	c.client.Disconnect()
}

func (c *appCollector) handleUplink(client mqtt.Client, appEUI types.AppEUI, devEUI types.DevEUI, req core.DataUpAppReq) {
	ctx := c.ctx.WithField("devEUI", devEUI)

	fields, valid, err := c.functions.Process(req.Payload)
	if err != nil {
		ctx.WithError(err).Error("Failed to process payload")
		return
	}
	if !valid {
		ctx.Warn("The payload is not valid")
		return
	}

	t, err := time.Parse(time.RFC3339, req.Metadata[0].ServerTime)
	if err != nil {
		ctx.WithError(err).Warnf("Invalid time: %v", req.Metadata[0].ServerTime)
		return
	}

	err = c.storage.Save(appEUI, devEUI, t, fields)
	if err != nil {
		ctx.WithError(err).Error("Failed to save data")
		return
	}
	ctx.Debug("Saved uplink packet in store")
}
{'fraction_non_alphanumeric': 0.13, 'fraction_numerical': 0.004347826086956522, 'mean_word_length': 4.050493962678376, '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': '6062765', 'n_tokens_mistral': 1680, 'n_tokens_neox': 1607, 'n_words': 416}