config/config.go

@@ -1,7 +1,6 @@
 package config
 
 import (
-	"bufio"
 	"bytes"
 	"fmt"
 	"io"
@@ -81,11 +80,10 @@ func validPort(port string) bool {
 }
 
 // ReadConfigFile takes two argument: folder and fileName.
-// Create if not exist
 // It reads its content, trims\n and \r, and return []byte
 // All errors are Fatal.
 func ReadConfigFile(folder string, fileName string) []byte {
-	f, err := os.OpenFile(folder+"/"+fileName, os.O_RDONLY|os.O_CREATE, 0666)
+	f, err := os.OpenFile("./"+folder+"/"+fileName, os.O_RDWR|os.O_CREATE, 0666)
 	defer f.Close()
 	if err != nil {
 		log.Fatal(err)
@@ -105,26 +103,3 @@ func ReadConfigFile(folder string, fileName string) []byte {
 	r := bytes.TrimSuffix(data[:count], []byte("\n"))
 	return bytes.TrimSuffix(r, []byte("\r"))
 }
-
-// ReadConfigFileLines takes two argument: folder and fileName.
-// Create if not exist
-// It reads its content line by line,
-// and return [][]byte
-// All errors are Fatal.
-func ReadConfigFileLines(folder string, fileName string) [][]byte {
-	res := [][]byte{}
-	f, err := os.OpenFile(folder+"/"+fileName, os.O_RDONLY|os.O_CREATE, 0666)
-    if err != nil {
-        log.Fatal(err)
-    }
-    defer f.Close()
-    scanner := bufio.NewScanner(f)
-    for scanner.Scan() {
-        res = append(res, []byte(scanner.Text()))
-    }
-
-    if err := scanner.Err(); err != nil {
-        log.Fatal(err)
-    }
-    return res
-}

d4-golang-utils_test.go

@@ -1,7 +1,9 @@
-package config
+package main
 
 import (
 	"testing"
+
+	config "github.com/D4-project/d4-golang-utils/config"
 )
 
 var testCases = []struct {
@@ -27,7 +29,7 @@ var testCases = []struct {
 func TestIsNet(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			b, _ := IsNet(tc.str)
+			b, _ := config.IsNet(tc.str)
 			if b != tc.expected {
 				t.Fail()
 			}

go.mod

@@ -2,12 +2,4 @@ module github.com/D4-project/d4-golang-utils
 
 go 1.12
 
-require (
+require github.com/gofrs/uuid v3.2.0+incompatible
-	github.com/gofrs/uuid v3.2.0+incompatible
-	github.com/gomodule/redigo v2.0.0+incompatible
-	github.com/olekukonko/tablewriter v0.0.5 // indirect
-	github.com/redislabs/redisgraph-go v2.0.2+incompatible
-	github.com/rjeczalik/notify v0.9.2
-	github.com/robfig/cron/v3 v3.0.1
-	github.com/stretchr/testify v1.7.0 // indirect
-)

go.sum

@@ -1,27 +1,2 @@
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
 github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
-github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
-github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
-github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0=
-github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
-github.com/olekukonko/tablewriter v0.0.5 h1:P2Ga83D34wi1o9J6Wh1mRuqd4mF/x/lgBS7N7AbDhec=
-github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6HuIJcUGPhkA7kY=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/redislabs/redisgraph-go v2.0.2+incompatible h1:HW9BqUhvgHeKXnUCrv9HOLPL/QFYxxzR2hsJClTzWk8=
-github.com/redislabs/redisgraph-go v2.0.2+incompatible/go.mod h1:GYn4oUFkbkHx49xm2H4G8jZziqKDKdRtDUuTBZTmqBE=
-github.com/rjeczalik/notify v0.9.2 h1:MiTWrPj55mNDHEiIX5YUSKefw/+lCQVoAFmD6oQm5w8=
-github.com/rjeczalik/notify v0.9.2/go.mod h1:aErll2f0sUX9PXZnVNyeiObbmTlk5jnMoCa4QEjJeqM=
-github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
-github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
-github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-golang.org/x/sys v0.0.0-20180926160741-c2ed4eda69e7 h1:bit1t3mgdR35yN0cX0G8orgLtOuyL9Wqxa1mccLB0ig=
-golang.org/x/sys v0.0.0-20180926160741-c2ed4eda69e7/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
-gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

inputreader/filewatcherreader.go

@@ -1,172 +0,0 @@
-package inputreader
-
-import (
-	"bytes"
-	"encoding/base64"
-	"fmt"
-	"github.com/rjeczalik/notify"
-	"github.com/robfig/cron/v3"
-	"io"
-	"log"
-	"os"
-	"time"
-)
-
-// FileWatcherReader is an abstraction of a folder watcher
-// and behaves like a reader
-type FileWatcherReader struct {
-	// Folder to watch
-	folderstr string
-	// Notify Channel
-	eic chan notify.EventInfo
-	// TearDown channel
-	exit chan string
-	// Chan used to restart the watching channel on a new folder
-	dailySwitch     chan bool
-	dailySwitchExit chan bool
-	// Current buffer
-	json bool
-	// Current file
-	curfile *os.File
-	// Current state Watching / Reading
-	watching bool
-	// Insert Separator
-	insertsep bool
-	// logging
-	log * log.Logger
-}
-
-// NewFileWatcherReader creates a new FileWatcherReader
-// json specifies whether we now we handle json files
-func NewFileWatcherReader(f string, j bool, daily bool, logger *log.Logger) (*FileWatcherReader, error) {
-	r := &FileWatcherReader{
-		folderstr:       f,
-		eic:             make(chan notify.EventInfo, 4096),
-		dailySwitch:     make(chan bool),
-		dailySwitchExit: make(chan bool),
-		json:            j,
-		watching:        true,
-		insertsep:       false,
-		log: logger,
-	}
-	// go routine holding the watcher
-	go setUpWatcher(r, daily)
-
-	// cron task to add daily folder to watch
-	if daily {
-		c := cron.New()
-		c.AddFunc("@midnight", func() {
-		//c.AddFunc("@every 10s", func() {
-			// Sending exit signal to setUpWatcher
-			r.dailySwitch <- true
-			// Waiting for exit signal from setUpWatcher
-			<-r.dailySwitchExit
-			go setUpWatcher(r, daily)
-		})
-		c.Start()
-	}
-	return r, nil
-}
-
-// setUpWatcher holds the watcher
-func setUpWatcher(r *FileWatcherReader, daily bool) {
-	if daily {
-		// TODO make it customizable
-		t, _ := time.ParseDuration("1s")
-		retryWatch(r, t)
-	} else {
-		if err := notify.Watch(fmt.Sprintf("%s/...", r.folderstr), r.eic, notify.InCloseWrite); err != nil {
-			log.Fatal(err)
-		}
-	}
-	defer notify.Stop(r.eic)
-	<-r.dailySwitch
-	r.dailySwitchExit <- true
-}
-
-// retryWatch tries to set up the watcher until it works every t
-func retryWatch(r *FileWatcherReader, t time.Duration) {
-	dt := time.Now()
-	//Format YYYYMMDD
-	// TODO make it customizable
-	currentFolder := dt.Format("20060102")
-	r.log.Println(fmt.Sprintf("Watching: %s/%s/...", r.folderstr, currentFolder))
-	for {
-		if err := notify.Watch(fmt.Sprintf("%s/%s/...", r.folderstr, currentFolder), r.eic, notify.InCloseWrite); err != nil {
-			r.log.Println(fmt.Sprintf("Waiting for: %s/%s/... to exist", r.folderstr, currentFolder))
-			time.Sleep(t)
-		}else{
-			return
-		}
-	}
-}
-
-// Read  waits for InCloseWrite file event uses a bytes reader to copy
-// the resulting file encoded in b64 in p
-func (fw *FileWatcherReader) Read(p []byte) (n int, err error) {
-
-	// Watching for new files to read
-	if fw.watching {
-	watchloop:
-		for {
-			select {
-			case ei := <-fw.eic:
-				//r.log.Println("Got event:", ei)
-				// New File, let's read its content
-				var err error
-				fw.curfile, err = os.Open(ei.Path())
-				if err != nil {
-					fw.log.Fatal(err)
-				}
-				fw.watching = false
-				break watchloop
-			case <-fw.exit:
-				// Exiting
-				return 0, io.EOF
-			}
-		}
-	}
-
-	// Inserting separator
-	if fw.insertsep {
-		var buf []byte
-		buf = append(buf, "\n"...)
-		rreader := bytes.NewReader(buf)
-		n, err = rreader.Read(p)
-		fw.watching = true
-		fw.insertsep = false
-		fw.curfile.Close()
-		return n, err
-	}
-
-	// Reading
-	// if not json it could be anything so we encode it in b64
-	if !fw.json {
-		// base64 stream encoder
-		b64buffer := new(bytes.Buffer)
-		b64encoder := base64.NewEncoder(base64.StdEncoding, b64buffer)
-		// max buffer size is then 3072 = 4096/4*3
-		buf := make([]byte, 3072)
-		bytesread, err := fw.curfile.Read(buf)
-		// buf is the input
-		b64encoder.Write(buf[:bytesread])
-		// Close the encoder to flush partially written blocks
-		b64encoder.Close()
-		if err == io.EOF {
-			fw.insertsep = true
-		}
-		// Copy from b64buffer to p
-		n, err = b64buffer.Read(p[:len(b64buffer.Bytes())])
-	} else {
-		n, err = fw.curfile.Read(p)
-		if err == io.EOF {
-			fw.insertsep = true
-		}
-	}
-	return n, err
-}
-
-// Teardown is called on error to stop the Reading loop if needed
-func (rl *FileWatcherReader) Teardown() {
-	rl.exit <- "exit"
-}

inputreader/redisreader.go

@@ -1,61 +0,0 @@
-package inputreader
-
-import (
-	"bytes"
-	"github.com/gomodule/redigo/redis"
-	"io"
-)
-
-// RedisLPOPReader is a abstraction of LPOP list
-// and behaves like a reader
-type RedisLPOPReader struct {
-	// D4 redis connection
-	r *redis.Conn
-	// D4 redis database
-	d int
-	// D4 Queue storing
-	q string
-	// Current buffer
-	buf []byte
-}
-
-// NewLPOPReader creates a new RedisLPOPReader
-func NewLPOPReader(rc *redis.Conn, db int, queue string) (*RedisLPOPReader, error) {
-	rr := *rc
-
-	if _, err := rr.Do("SELECT", db); err != nil {
-		rr.Close()
-		return nil, err
-	}
-
-	r := &RedisLPOPReader{
-		r: rc,
-		d: db,
-		q: queue,
-	}
-
-	return r, nil
-}
-
-// Read LPOP the redis queue and use a bytes reader to copy
-// the resulting data in p
-func (rl *RedisLPOPReader) Read(p []byte) (n int, err error) {
-	rr := *rl.r
-
-	buf, err := redis.Bytes(rr.Do("LPOP", rl.q))
-	// If redis return empty: EOF (user should not stop)
-	if err == redis.ErrNil {
-		return 0, io.EOF
-	} else if err != nil {
-		return 0, err
-	}
-	buf = append(buf, "\n"...)
-	rreader := bytes.NewReader(buf)
-	n, err = rreader.Read(p)
-	return n, err
-}
-
-// Teardown is called on error to close the redis connection
-func (rl *RedisLPOPReader) Teardown() {
-	(*rl.r).Close()
-}