package main

import (
	"fmt"
	streams "gitee.com/qw_1215/glink"
	si "gitee.com/qw_1215/glink/extension/sink"
	so "gitee.com/qw_1215/glink/extension/source"
	"gitee.com/qw_1215/glink/flow"
	"strings"

	"github.com/apache/pulsar-client-go/pulsar"
)

func main() {
	clientOptions := pulsar.ClientOptions{URL: "pulsar://172.17.126.201:6650"}
	producerOptions := pulsar.ProducerOptions{Topic: "test2"}
	consumerOptions := pulsar.ConsumerOptions{
		Topic:            "test1",
		SubscriptionName: "group1",
		Type:             pulsar.Exclusive,
	}

	source, err := so.NewPulsarSource(&clientOptions, &consumerOptions)
	streams.Check(err)
	flow1 := flow.NewMap(toUpper, 1)
	sink := si.NewPulsarSink(&clientOptions, &producerOptions)

	source.Via(flow1).To(sink)
}

var toUpper = func(in interface{}) interface{} {
	msg := in.(pulsar.Message)
	mm := strings.ToUpper(string(msg.Payload()))
	fmt.Println(mm)
	return mm
}