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"
)

// Test producer: nc -u 127.0.0.1 3434
// Test 订阅: nc -u -l 3535
func main() {
	source, err := so.NewNetSource(so.UDP, "127.0.0.1:3434")
	streams.Check(err)
	flow1 := flow.NewMap(toUpper, 1)
	sink, err := si.NewNetSink(si.ConnType(so.UDP), "127.0.0.1:3535")
	streams.Check(err)

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

var toUpper = func(in interface{}) interface{} {
	msg := in.(string)
	fmt.Printf("Got: %s\n", msg)
	return strings.ToUpper(msg)
}