<commit_before>package main

import (
	"github.com/garyburd/redigo/redis"
	"time"
	"gopkg.in/olivere/elastic.v3"

	"encoding/json"
	"log"
	"fmt"
)

var redisChan chan Candy

type Candy struct {
	Name, Object string
	Price        float64
	Time         time.Time
}

func redisSubscriber() {
	conn, err := redis.Dial("tcp", "redis:6379")
	if err != nil {
		panic("I don't want to live on this planet anymore")
	}
	psc := redis.PubSubConn{conn}
	psc.Subscribe("candy")
	for {
		switch v := psc.Receive().(type) {
		case redis.Message:
			fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
			if v.Channel == "candy" {
				var c Candy
				err := json.Unmarshal(v.Data, c)
				if (err != nil) {
					log.Printf("Seems our redis is sick! In the evening we'll get some schnaps to ease the pain!")
				}
				redisChan <- c
			}
		case redis.Subscription:
			fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
		case error:
			log.Println(v)
		}
	}
}

func elasticUpdater() {
	client, err := elastic.NewClient(elastic.SetURL("http://elasticsearch:9200"))
	if err != nil {
		// Handle error
	}

	_, err = client.CreateIndex("twitter").Do()
	if err != nil {
		// Handle error
		panic(err)
	}

	// Add a document to the index

	for c := range redisChan {
		_, err = client.Index().
		Index("candyshop").
		Type("candy").
		BodyJson(c).
		Do()
		if err != nil {
			// Handle error
			log.Println(err)
		}
	}
}

func main() {
	redisChan = make(chan Candy, 10)
	go redisSubscriber()
	elasticUpdater()


}<commit_msg>removed elastic search lib and use http post<commit_after>package main

import (
	"github.com/garyburd/redigo/redis"
	"time"

	"encoding/json"
	"log"
	"fmt"
	"net/http"
)

var redisChan chan Candy

type Candy struct {
	Name, Object string
	Price        float64
	Time         time.Time
}

func redisSubscriber() {
	conn, err := redis.Dial("tcp", "redis:6379")
	if err != nil {
		panic("I don't want to live on this planet anymore")
	}
	psc := redis.PubSubConn{conn}
	psc.Subscribe("candy")
	for {
		switch v := psc.Receive().(type) {
		case redis.Message:
			fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
			if v.Channel == "candy" {
				var c Candy
				err := json.Unmarshal(v.Data, c)
				if (err != nil) {
					log.Printf("Seems our redis is sick! In the evening we'll get some schnaps to ease the pain!")
				}
				redisChan <- c
			}
		case redis.Subscription:
			fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
		case error:
			log.Println(v)
		}
	}
}

func elasticUpdater() {


	// Add a document to the index

	for c := range redisChan {
		buf, err := json.Marshal(c)
		if err != nil {
			log.Print(err)
			continue
		}
		resp, err := http.Post("http://elasticsearch:9200/candyshop/candy/", "application/json", &buf)
		if err != nil {
			log.Println(err)
		}
		if resp != nil {
			log.Println(resp)
		}
	}
}

func main() {
	time.Sleep(time.Second * 20)
	redisChan = make(chan Candy, 10)
	go redisSubscriber()
	elasticUpdater()


}
{'fraction_non_alphanumeric': 0.153507285665876, 'fraction_numerical': 0.009149440867502542, 'mean_word_length': 2.96241610738255, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5166753', 'n_tokens_mistral': 1300, 'n_tokens_neox': 1156, 'n_words': 292}