Merge pull request #26 from DudeWhoCode/version/0.1.2

Version/0.1.2

Author: DudeWhoCode

.gitignore

@@ -60,3 +60,6 @@ crashlytics-build.properties
 fabric.properties
 *.zip
 *.txt
+
+# python
+env/

.travis.yml

@@ -0,0 +1,40 @@
+dist: trusty
+
+sudo: required
+
+services:
+  - redis-server
+
+language: go
+
+os:
+    - linux
+
+go:
+  - 1.8
+
+python:
+  - 2.7
+
+env:
+   - KULAY_CONF=$TRAVIS_BUILD_DIR/testdata/kulay.toml BOTO_CONFIG=/dev/null
+
+install:
+  - pip install moto
+  - pip install awscli
+  - pip install flask
+
+before_install:
+   - go get -u github.com/kardianos/govendor
+   - govendor sync
+
+before_script:
+    - sh testdata/prepare.sh
+
+script:
+    - sh test.sh
+
+after_success:
+    - bash <(curl -s https://codecov.io/bash)
+
+

README.md

@@ -1,7 +1,9 @@
 # Kulay
+[![Build Status](https://travis-ci.org/DudeWhoCode/kulay.svg?branch=master)](https://travis-ci.org/DudeWhoCode/kulay)
+[![codecov](https://codecov.io/gh/DudeWhoCode/kulay/branch/master/graph/badge.svg)](https://codecov.io/gh/DudeWhoCode/kulay)
+[![Go Report Card](https://goreportcard.com/badge/github.com/dudewhocode/kulay)](https://goreportcard.com/report/github.com/dudewhocode/kulay)
 [![GitHub license](https://img.shields.io/badge/license-MIT-blue.svg)](https://raw.githubusercontent.com/DudeWhoCode/kulay/master/LICENSE)
 [![GitHub issues](https://img.shields.io/github/issues/DudeWhoCode/kulay.svg)](https://github.com/DudeWhoCode/kulay/issues)
-[![Go Report Card](https://goreportcard.com/badge/github.com/dudewhocode/kulay)](https://goreportcard.com/report/github.com/dudewhocode/kulay)
 
 An high speed message passing system between various queues and services.
 
@@ -65,11 +67,72 @@ Each sub section can be any of your favourite name strictly preceeded by the sec
 ```
 Here the logbuffer is a custom name you give to your subsection, it can be production_buffer or prodqueue or which ever makes more sense to you. The `host`, `port`, `password`, `database` are the options we need to initilize the redis client and `queue` is the key name of the list in redis which will be used as queue.
 
+Currently kulay supports passing messages between :
+1. Redis queue
+2. Redis pubsub
+3. AWS SQS
+4. Jsonl file read/write
+
+### Config structures for supported services : 
+#### Redis pubsub
+```
+[redispubsub]
+       [redispubsub.yourCustomName]
+       host = "localhost"
+       port = "6379"
+       password = "topsecret"
+       database = 0
+       channel = "mychannel"
+```
+host - Your redis hostname or IP address   
+port - Port in which redis runs, default is 6379   
+password - Password for your redis server, leave it as "" for no password     
+channel - The pubsub channel which you need to send or receive messages 
+ 
+
+### Redis queue
+```
+[redisq]
+       [redisq.yourCustomName]
+       host = "localhost"
+       port = "6379"
+       password = "topsecret"
+       database = 0
+       queue = "mychannel"
+```
+host - Your redis hostname or IP address   
+port - Port in which redis runs, default is 6379   
+password - Password for your redis server, leave it as "" for no password        
+queue - The queue to which you will send or receive messages   
+
+### SQS
+```
+[sqs]
+    [sqs.test_singapore]
+    queue_url = "https://sqs.ap-southeast-1.amazonaws.com/12345678/test_queue"
+    region = "ap-southeast-1"
+    delete_msg = true
+```
+queue_url - URL of the queue found in AWS console    
+region - The region where given queue was created   
+delete_msg - Delete flag, should be true if you want to delete the message from sqs after reading      
+
+### Jsonl
+```
+jsonl]
+    [jsonl.local_backup]
+    path = "/tmp/backup.jsonl"
+    rotate = true
+    batch = 1000
+```
+path   - Location where the files has to be created  
+rotate - If rotate flag is enabled, kulay will create a new file if the lines crosses the count in `batch` field    
+batch  - The line count for a single file if rotate=true   
+  
 # Built with
 * [Cobra](https://github.com/spf13/cobra) - command line framework
 * [Viper](https://github.com/spf13/viper) - configuration handler
 * [Logrus](https://github.com/sirupsen/logrus) - logging
 
 # Versioning
-This project uses [SemVer](http://semver.org/) for versioning. For the versions available, see the [tags on this repository](https://github.com/DudeWhoCode/kulay/tags)
-
+This project uses [SemVer](http://semver.org/) for versioning. For the versions available, see the [tags on this repository](https://github.com/DudeWhoCode/kulay/tags)

backend/fileio/jsonl.go

@@ -2,17 +2,59 @@ package fileio
 
 import (
 	"bufio"
+	"fmt"
 	. "github.com/DudeWhoCode/kulay/logger"
 	"os"
+	"path/filepath"
+	"strings"
+	"time"
 )
 
-func Put(fpath string, rec <-chan string) {
-	f, err := os.Create(fpath)
+type rotateFile struct {
+	file  string
+	ext   string
+	count int
+}
+
+func initRotate(fpath string) *rotateFile {
+	count := 0
+	ext := filepath.Ext(fpath)
+	file := strings.TrimSuffix(fpath, ext)
+	return &rotateFile{
+		file,
+		ext,
+		count,
+	}
+}
+
+func (f *rotateFile) newFile() (file string) {
+	now := time.Now()
+	nanos := now.UnixNano()
+	file = fmt.Sprintf("%s-%d-%d%s", f.file, nanos, f.count, f.ext)
+	f.count++
+	return
+}
+
+func Put(fpath string, batch int, rec <-chan string, rotate bool) {
+	r := initRotate(fpath)
+	chanCnt := 0
+	newFileName := r.newFile()
+	f, err := os.Create(newFileName)
+	fmt.Println("created : ", newFileName)
 	if err != nil {
-		Log.Error("Unable to open file for writing jsonl")
+		Log.Error("Unable to open file for writing jsonl\n", err)
 	}
 	for msg := range rec {
+		if rotate == true && chanCnt != 0 && chanCnt%batch == 0 {
+			f.Close()
+			newFileName := r.newFile()
+			f, err = os.Create(newFileName)
+			if err != nil {
+				Log.Error("Unable to open file for writing jsonl")
+			}
+		}
 		f.Write([]byte(msg + "\n"))
+		chanCnt++
 	}
 	defer f.Close()
 }

backend/fileio/jsonl_test.go

@@ -4,12 +4,20 @@ import (
 	"bufio"
 	"encoding/json"
 	"os"
+	"path/filepath"
+	"strings"
 	"testing"
 )
 
 func TestPut(t *testing.T) {
-	fpath := "/tmp/test_put.jsonl"
+	path := "/tmp/jsonltest/"
+	os.RemoveAll(path)
+	os.MkdirAll(path, 0777)
+	fpath := path + "test_put.jsonl"
 	pipe := make(chan string)
+	batch := 5
+	rotate := false
+	testCnt := 10
 	type test struct {
 		Name  string `json:"name"`
 		Desc  string `json:"desc"`
@@ -23,10 +31,16 @@ func TestPut(t *testing.T) {
 		135,
 	}
 	testStr, _ := json.Marshal(testData)
-	go Put(fpath, pipe)
-	for i := 1; i <= 10; i++ {
+	go Put(fpath, batch, pipe, rotate)
+	for i := 1; i <= testCnt; i++ {
 		pipe <- string(testStr)
 	}
+	fileList := []string{}
+	filepath.Walk(path, func(path string, f os.FileInfo, err error) error {
+		fileList = append(fileList, path)
+		return nil
+	})
+	fpath = fileList[1]
 	file, err := os.Open(fpath)
 	if err != nil {
 		t.Errorf("Expected no errors in reading file, got %v", err)
@@ -44,6 +58,58 @@ func TestPut(t *testing.T) {
 	}
 }
 
+func TestPutRotate(t *testing.T) {
+	path := "/tmp/jsonltest/"
+	os.RemoveAll(path)
+	os.MkdirAll(path, 0777)
+	fpath := path + "test_put.jsonl"
+	pipe := make(chan string)
+	batch := 5
+	rotate := true
+	testCnt := 20
+	type test struct {
+		Name  string `json:"name"`
+		Desc  string `json:"desc"`
+		Url   string `json:"url"`
+		Stars int    `json:"stars"`
+	}
+	testData := &test{
+		"kulay",
+		"High speed message routing between services",
+		"https://github.com/kulay",
+		135,
+	}
+	testStr, _ := json.Marshal(testData)
+	go Put(fpath, batch, pipe, rotate)
+	for i := 1; i <= testCnt; i++ {
+		pipe <- string(testStr)
+	}
+	fileList := []string{}
+	filepath.Walk(path, func(path string, f os.FileInfo, err error) error {
+		fileList = append(fileList, path)
+		return nil
+	})
+	for _, fpath := range fileList {
+		if !strings.HasSuffix(fpath, "jsonl") {
+			continue
+		}
+		file, err := os.Open(fpath)
+		if err != nil {
+			t.Errorf("Expected no errors in reading file, got %v", err)
+		}
+		scanner := bufio.NewScanner(file)
+		for scanner.Scan() {
+			if err := json.Unmarshal(scanner.Bytes(), &test{}); err != nil {
+				t.Errorf("Expected no errors in unmarshalling jsonline, got %v", err)
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			t.Errorf("Expected no errors in scanning file, got %v", err)
+		}
+		file.Close()
+	}
+}
+
 func TestGet(t *testing.T) {
 	fpath := "/tmp/test_get.jsonl"
 	testCnt := 10

backend/redispubsub/redispubsub.go

@@ -0,0 +1,30 @@
+package redispubsub
+
+import (
+	"github.com/DudeWhoCode/kulay/backend"
+	. "github.com/DudeWhoCode/kulay/logger"
+)
+
+func Get(host string, port string, pass string, db int, channel string, rec chan<- string) {
+	client := backend.NewRedisSession(host, port, pass, db)
+	pubsub := client.Subscribe(channel)
+	for {
+		msg, err := pubsub.ReceiveMessage()
+		if err != nil {
+			Log.Error("Unable to receive message from redis pubsub channel\n", err)
+		}
+		if msg.Payload == "$^KILL^$" {
+			break
+		}
+		rec <- msg.Payload
+	}
+}
+
+func Put(host string, port string, pass string, db int, channel string, snd <-chan string) {
+	client := backend.NewRedisSession(host, port, pass, db)
+	for msg := range snd {
+		if err := client.Publish(channel, msg).Err(); err != nil {
+			Log.Error("Unable to publish message tp redis pubsub channel\n", err)
+		}
+	}
+}