Merge pull request #46 from JPercivall/Adding_raw_socket_producer

Adding a producer to send over raw TCP and UDP

Author: mehrdadrad

.gitignore

@@ -0,0 +1,9 @@
+stress/
+vflow/
+.DS_Store
+
+# Intellij
+.idea/
+*.iml
+*.iws
+*~

docs/config.md

@@ -134,3 +134,26 @@ The NATS configuration contains the following key
 |Key                  | Default               |  Environment variable    | Description                                                      |
 |---------------------| ----------------------|--------------------------|------------------------------------------------------------------|
 |url                  | nats://localhost:4222 | NA                       | URL addresse     
+
+# Raw Socket Configuration
+
+Note that for messages sent over TCP and UDP using this producer, the message deliminator is a new line character ("\n").
+
+## Format
+A config file is a plain text file in [YAML](https://en.wikipedia.org/wiki/YAML) format.
+
+```
+key: value
+```
+
+The default configuration file is /etc/vflow/mq.conf, you can be able to change it through vFlow configuration.
+
+
+## Configuration Keys
+The NATS configuration contains the following key
+
+|Key                  | Default               |  Environment variable    | Description                                                          |
+|---------------------| ----------------------|--------------------------|----------------------------------------------------------------------|
+|url                  | localhost:9555        | NA                       | URL address to send to. Includes the hostname and port.              |     
+|protocol             | tcp                   | NA                       | Protocol to use to send. Can be either "tcp" or "udp"                |
+|retry-max            | 2                     | NA                       | The number of times a message will be retried before giving up on it |     

producer/producer.go

@@ -48,9 +48,10 @@ type MQueue interface {
 // NewProducer constructs new Messaging Queue
 func NewProducer(mqName string) *Producer {
 	var mqRegistered = map[string]MQueue{
-		"kafka": new(Kafka),
-		"nsq":   new(NSQ),
-		"nats":  new(NATS),
+		"kafka":     new(Kafka),
+		"nsq":       new(NSQ),
+		"nats":      new(NATS),
+		"rawSocket": new(RawSocket),
 	}
 
 	return &Producer{

producer/rawSocket.go

@@ -0,0 +1,129 @@
+//: ----------------------------------------------------------------------------
+//: Copyright (C) 2017 Verizon.  All Rights Reserved.
+//: All Rights Reserved
+//:
+//: file:    rawSocket.go
+//: details: vflow tcp/udp producer plugin
+//: author:  Joe Percivall
+//: date:    12/18/2017
+//:
+//: Licensed under the Apache License, Version 2.0 (the "License");
+//: you may not use this file except in compliance with the License.
+//: You may obtain a copy of the License at
+//:
+//:     http://www.apache.org/licenses/LICENSE-2.0
+//:
+//: Unless required by applicable law or agreed to in writing, software
+//: distributed under the License is distributed on an "AS IS" BASIS,
+//: WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//: See the License for the specific language governing permissions and
+//: limitations under the License.
+//: ----------------------------------------------------------------------------
+
+package producer
+
+import (
+	"io/ioutil"
+	"log"
+	"strings"
+
+	"fmt"
+	"gopkg.in/yaml.v2"
+	"net"
+)
+
+// RawSocket represents RawSocket producer
+type RawSocket struct {
+	connection net.Conn
+	config     RawSocketConfig
+	logger     *log.Logger
+}
+
+// RawSocketConfig is the struct that holds all configuation for RawSocketConfig connections
+type RawSocketConfig struct {
+	URL      string `yaml:"url"`
+	Protocol string `yaml:"protocol"`
+	MaxRetry int    `yaml:"retry-max"`
+}
+
+func (rs *RawSocket) setup(configFile string, logger *log.Logger) error {
+	var err error
+	rs.config = RawSocketConfig{
+		URL:      "localhost:9555",
+		Protocol: "tcp",
+		MaxRetry: 2,
+	}
+
+	if err = rs.load(configFile); err != nil {
+		logger.Println(err)
+		return err
+	}
+
+	rs.connection, err = net.Dial(rs.config.Protocol, rs.config.URL)
+	if err != nil {
+		logger.Println(err)
+		return err
+	}
+
+	rs.logger = logger
+
+	return nil
+}
+
+func (rs *RawSocket) inputMsg(topic string, mCh chan []byte, ec *uint64) {
+	var (
+		msg []byte
+		err error
+		ok  bool
+	)
+
+	rs.logger.Printf("start producer: RawSocket, server: %+v, Protocol: %s\n",
+		rs.config.URL, rs.config.Protocol)
+
+	for {
+		msg, ok = <-mCh
+		if !ok {
+			break
+		}
+
+		for i := 0; ; i++ {
+			_, err = fmt.Fprintf(rs.connection, string(msg)+"\n")
+			if err == nil {
+				break
+			}
+
+			*ec++
+
+			if strings.HasSuffix(err.Error(), "broken pipe") {
+				var newConnection, err = net.Dial(rs.config.Protocol, rs.config.URL)
+				if err != nil {
+					rs.logger.Println("Error when attempting to fix the broken pipe", err)
+				} else {
+					rs.logger.Println("Successfully reconnected")
+					rs.connection = newConnection
+				}
+			}
+
+			if i >= (rs.config.MaxRetry) {
+				rs.logger.Println("message failed after the configured retry limit:", err)
+				break
+			} else {
+				rs.logger.Println("retrying after error:", err)
+			}
+		}
+	}
+}
+
+func (rs *RawSocket) load(f string) error {
+	b, err := ioutil.ReadFile(f)
+	if err != nil {
+		return err
+	}
+
+	err = yaml.Unmarshal(b, &rs.config)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}