producer.go

package producer

import (
	"config"
	"fmt"

	"os"
	"time"

	"github.com/aws/aws-sdk-go/service/kinesis"

	mqtt "github.com/eclipse/paho.mqtt.golang"
)

func postDataTokinesisStream(client mqtt.Client, message mqtt.Message) {
	fmt.Printf("Received message on topic: %s\nMessage: %s\n", message.Topic(), message.Payload())
	streamName := config.GetKinesisStreamName()
	kclient := config.GetKinesisClient()
	var putRecordInput kinesis.PutRecordInput
	partitionKey := message.Topic()
	putRecordInput.PartitionKey = &partitionKey
	putRecordInput.StreamName = &streamName
	putRecordInput.Data = message.Payload()
	putRecordOutput, err := kclient.PutRecord(&putRecordInput)
	if err != nil {
		fmt.Println(err)
	} else {
		fmt.Println(putRecordOutput)
	}

}

func StartKinesisProducer() {
	fmt.Println("Kinesis Producer Started")
	c := make(chan os.Signal, 1)
	opts := mqtt.NewClientOptions().AddBroker(config.GetMqttServerurl()).SetClientID("MqttSubscriberClient")
	opts.SetKeepAlive(2 * time.Second)
	opts.SetPingTimeout(1 * time.Second)
	opts.OnConnect = func(c mqtt.Client) {
		if token := c.Subscribe(config.GetMQTTTopicName(), 0, postDataTokinesisStream); token.Wait() && token.Error() != nil {
			panic(token.Error())
		}
	}

	client := mqtt.NewClient(opts)
	if token := client.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	} else {
		fmt.Printf("Connected to %s\n", config.GetMqttServerurl())
	}

	<-c
}