Merge pull request #20 from reugn/develop

v0.3.0

Author: reugn

.github/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,7 @@
+## Motivation
+* Fixes #___
+*
+## Modifications
+*
+## Verify change
+* [ ] Make sure the change passes the CI checks.

CONTRIBUTING.md

@@ -0,0 +1,26 @@
+# How to contribute
+
+If you would like to contribute code to this project, fork the repository and send a pull request.
+
+## Prerequisite
+
+If you have not installed Go, install it according to the [installation instruction](http://golang.org/doc/install).
+Since the `go mod` package management tool is used in this project, **Go 1.11 or higher** version is required.
+
+## Fork
+
+Before contributing, you need to fork [go-streams](https://github.com/reugn/go-streams) to your github repository.
+
+## Contribution flow
+
+* [Configure remote for a fork](https://help.github.com/en/github/collaborating-with-issues-and-pull-requests/configuring-a-remote-for-a-fork)
+* git checkout -b <your_branch>
+* git add .
+* git commit -m "commit message"
+* git push --set-upstream origin <your_branch>
+* Verify all tests and CI checks pass
+* [Create a pull request](https://help.github.com/en/github/collaborating-with-issues-and-pull-requests/creating-a-pull-request)
+
+## Code style
+
+The coding style suggested by the Golang community is used in `go-streams`. For details, refer to [style doc](https://github.com/golang/go/wiki/CodeReviewComments).

README.md

@@ -31,7 +31,9 @@ Flow capabilities (flow package):
 Supported Sources and Sinks (ext package):
 * Go channels
 * File system
-* [Kafka](https://kafka.apache.org/)
+* Network (TCP, UDP)
+* [Apache Kafka](https://kafka.apache.org/)
+* [Apache Pulsar](https://pulsar.apache.org/)
 * [Redis](https://redis.io/)
 
 ## Examples

examples/net/.gitignore

@@ -0,0 +1 @@
+net

examples/net/main.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/reugn/go-streams"
+	ext "github.com/reugn/go-streams/extension"
+	"github.com/reugn/go-streams/flow"
+)
+
+// Test producer: nc -u 127.0.0.1 3434
+// Test consumer: nc -u -l 3535
+func main() {
+	source, err := ext.NewNetSource(ext.UDP, "127.0.0.1:3434")
+	streams.Check(err)
+	flow1 := flow.NewMap(toUpper, 1)
+	sink, err := ext.NewNetSink(ext.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)
+}

examples/pulsar/.gitignore

@@ -0,0 +1 @@
+pulsar

examples/pulsar/docker-compose.yml

@@ -0,0 +1,24 @@
+version: '3'
+
+services:
+  standalone:
+    image: apachepulsar/pulsar
+    ports:
+      - "6650:6650"
+    expose:
+      - 8080
+      - 6650 
+    environment:
+      - PULSAR_MEM=" -Xms512m -Xmx512m -XX:MaxDirectMemorySize=1g"
+    command: >
+      /bin/bash -c
+      "bin/apply-config-from-env.py conf/standalone.conf
+      && bin/pulsar standalone"
+  dashboard:
+    image: apachepulsar/pulsar-dashboard
+    depends_on:
+      - standalone
+    ports:
+      - "80:80"
+    environment:
+      - SERVICE_URL=http://standalone:8080

examples/pulsar/main.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"strings"
+
+	"github.com/apache/pulsar-client-go/pulsar"
+	"github.com/reugn/go-streams"
+	ext "github.com/reugn/go-streams/extension"
+	"github.com/reugn/go-streams/flow"
+)
+
+func main() {
+	clientOptions := pulsar.ClientOptions{URL: "pulsar://localhost:6650"}
+	producerOptions := pulsar.ProducerOptions{Topic: "test2"}
+	consumerOptions := pulsar.ConsumerOptions{
+		Topic:            "test1",
+		SubscriptionName: "group1",
+		Type:             pulsar.Exclusive,
+	}
+
+	source, err := ext.NewPulsarSource(&clientOptions, &consumerOptions)
+	streams.Check(err)
+	flow1 := flow.NewMap(toUpper, 1)
+	sink := ext.NewPulsarSink(&clientOptions, &producerOptions)
+
+	source.Via(flow1).To(sink)
+}
+
+var toUpper = func(in interface{}) interface{} {
+	msg := in.(pulsar.Message)
+	return strings.ToUpper(string(msg.Payload()))
+}

extension/net.go

@@ -0,0 +1,167 @@
+package ext
+
+import (
+	"bufio"
+	"errors"
+	"log"
+	"net"
+	"time"
+
+	"github.com/reugn/go-streams"
+	"github.com/reugn/go-streams/flow"
+)
+
+// ConnType connection type
+type ConnType string
+
+const (
+	// TCP connection type
+	TCP ConnType = "tcp"
+	// UDP connection type
+	UDP ConnType = "udp"
+)
+
+// NetSource network socket connector
+type NetSource struct {
+	conn     net.Conn
+	listener net.Listener
+	connType ConnType
+	out      chan interface{}
+}
+
+// NewNetSource creates a new NetSource
+func NewNetSource(connType ConnType, address string) (*NetSource, error) {
+	var err error
+	var conn net.Conn
+	var listener net.Listener
+	out := make(chan interface{})
+
+	switch connType {
+	case TCP:
+		addr, _ := net.ResolveTCPAddr(string(connType), address)
+		listener, err = net.ListenTCP(string(connType), addr)
+
+		if err != nil {
+			log.Fatal(err)
+			return nil, err
+		}
+
+		go acceptConnections(listener, out)
+
+	case UDP:
+		addr, _ := net.ResolveUDPAddr(string(connType), address)
+		conn, err = net.ListenUDP(string(connType), addr)
+
+		if err != nil {
+			log.Fatal(err)
+			return nil, err
+		}
+
+		go handleConnection(conn, out)
+
+	default:
+		return nil, errors.New("Invalid connection type")
+	}
+
+	source := &NetSource{
+		conn:     conn,
+		listener: listener,
+		connType: connType,
+		out:      out,
+	}
+
+	return source, nil
+}
+
+// TCP Accept routine
+func acceptConnections(listener net.Listener, out chan<- interface{}) {
+	for {
+		// accept new connection
+		conn, err := listener.Accept()
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		// handle new connection
+		go handleConnection(conn, out)
+	}
+	log.Printf("Closing NetSource TCP listener %v", listener)
+	listener.Close()
+}
+
+// handleConnection routine
+func handleConnection(conn net.Conn, out chan<- interface{}) {
+	log.Printf("NetSource connected on: %v", conn.LocalAddr())
+	reader := bufio.NewReader(conn)
+	for {
+		bufferBytes, err := reader.ReadBytes('\n')
+		if err != nil {
+			log.Fatal(err)
+			break
+		} else {
+			out <- string(bufferBytes)
+		}
+	}
+	log.Printf("Closing NetSource connection %v", conn)
+	conn.Close()
+}
+
+// Via streams data through given flow
+func (ns *NetSource) Via(_flow streams.Flow) streams.Flow {
+	flow.DoStream(ns, _flow)
+	return _flow
+}
+
+// Out returns channel for sending data
+func (ns *NetSource) Out() <-chan interface{} {
+	return ns.out
+}
+
+// NetSink downstreams input events to a network soket
+type NetSink struct {
+	conn     net.Conn
+	connType ConnType
+	in       chan interface{}
+}
+
+// NewNetSink creates a new NetSink
+func NewNetSink(connType ConnType, address string) (*NetSink, error) {
+	var err error
+	var conn net.Conn
+
+	conn, err = net.DialTimeout(string(connType), address, time.Second*10)
+	if err != nil {
+		return nil, err
+	}
+
+	sink := &NetSink{
+		conn:     conn,
+		connType: connType,
+		in:       make(chan interface{}),
+	}
+
+	go sink.init()
+	return sink, nil
+}
+
+// start main loop
+func (ns *NetSink) init() {
+	log.Printf("NetSink connected on: %v", ns.conn.LocalAddr())
+	writer := bufio.NewWriter(ns.conn)
+	for msg := range ns.in {
+		switch m := msg.(type) {
+		case string:
+			_, err := writer.WriteString(m)
+			if err == nil {
+				err = writer.Flush()
+			}
+		}
+	}
+	log.Printf("Closing NetSink connection %v", ns.conn)
+	ns.conn.Close()
+}
+
+// In returns channel for receiving data
+func (ns *NetSink) In() chan<- interface{} {
+	return ns.in
+}