<commit_before>package workque

import (
	"log"
	"math/rand"
	"sync"
	"time"

	"github.com/Shopify/sarama"
)

var quit = make(chan *sync.WaitGroup, 1)

// Payload a Kafka consumer message
type Payload struct {
	Message sarama.ConsumerMessage
}

// Worker represents the worker that executes the job
type Worker struct {
	WorkerPool chan chan Job
	JobChannel chan Job
}

// NewWorker create new channel worker
func NewWorker(workerPool chan chan Job) Worker {
	return Worker{
		WorkerPool: workerPool,
		JobChannel: make(chan Job)}
}

// Start method starts the run loop for the worker,
// listening for a quit channel in case we need to stop it
func (w Worker) Start() {
	go func() {
		// register the current worker into the worker queue.
		w.WorkerPool <- w.JobChannel
		log.Println("Worker started...")

		for {
			select {
			case job := <-w.JobChannel:
				// we have received a work request.
				if err := job.Payload.InsertIntoDB(); err != nil {
					log.Printf("Error inserting into database: %s", err.Error())
				}

			case wg := <-quit:
				// we have received a signal to stop
				log.Println("Worker stopped")
				wg.Done()
				return
			}
		}
	}()
}

// Stop signals the worker to stop listening for work requests.
func (w Worker) Stop(wg *sync.WaitGroup) {
	go func() {
		log.Println("Stopping worker...")
		quit <- wg
	}()
}

// InsertIntoDB insert receive data into database
func (p *Payload) InsertIntoDB() error {
	log.Printf("worker process message - Topic: %s, Partition: %d, Offset: %d,\tMessageKey: %s,\tMessageValue: %s", p.Message.Topic, p.Message.Partition, p.Message.Offset, p.Message.Key, p.Message.Value)
	time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond)
	log.Println("work done")
	return nil
}<commit_msg> insert into real database, and fix  issue worker stop after process<commit_after>package workque

import (
	"encoding/json"
	"log"
	"sync"
	"time"

	"github.com/gocql/gocql"
)

var quit = make(chan *sync.WaitGroup, 1)

// Data a MQTT payload model
type Data struct {
	Temperature string `json:"temperature"`
	Humidity    string `json:"humidity"`
}

// Worker represents the worker that executes the job
type Worker struct {
	WorkerPool chan chan Job
	JobChannel chan Job
}

// NewWorker create new channel worker
func NewWorker(workerPool chan chan Job) Worker {
	return Worker{
		WorkerPool: workerPool,
		JobChannel: make(chan Job)}
}

// Start method starts the run loop for the worker,
// listening for a quit channel in case we need to stop it
func (w Worker) Start() {
	go func() {
		log.Println("Worker started")

		for {
			// register the current worker into the worker queue.
			w.WorkerPool <- w.JobChannel

			select {
			case job := <-w.JobChannel:
				// we have received a work request.
				if err := job.InsertIntoDB(); err != nil {
					log.Printf("Error inserting into database: %s", err.Error())
				}

			case wg := <-quit:
				// we have received a signal to stop
				log.Println("Worker stopped")
				wg.Done()
				return
			}
		}
	}()
}

// Stop signals the worker to stop listening for work requests.
func (w Worker) Stop(wg *sync.WaitGroup) {
	go func() {
		log.Println("Stopping worker...")
		quit <- wg
	}()
}

// InsertIntoDB insert received data into database
func (j *Job) InsertIntoDB() error {
	log.Printf("worker process message - Topic: %s, Partition: %d, Offset: %d,\tMessageKey: %s,\tMessageValue: %s", j.Message.Topic, j.Message.Partition, j.Message.Offset, j.Message.Key, j.Message.Value)
	var data Data
	if err := json.Unmarshal(j.Message.Value, &data); err != nil {
		return err
	}

	if err := j.DB.Query(`INSERT INTO data (id, username, temperature, humidity, timestamp) VALUES (?, ?, ?, ?, ?)`,
		gocql.TimeUUID(), j.Message.Key, data.Temperature, data.Humidity, time.Now()).Exec(); err != nil {
		return err
	}
	log.Println("work done")
	return nil
}
{'fraction_non_alphanumeric': 0.12835199166883624, 'fraction_numerical': 0.0013017443374121322, 'mean_word_length': 3.4622531939605112, 'pattern_counts': {'":': 0, '<': 11, '<?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': '7532676', 'n_tokens_mistral': 1388, 'n_tokens_neox': 1248, 'n_words': 420}