queue complete

kafka.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net"
 	"sync/atomic"
+	"time"
 
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
@@ -68,8 +69,22 @@ func (w *Worker) Queue(job core.QueuedMessage) (err error) {
 	if atomic.LoadInt32(&w.stopFlag) == 1 {
 		return queue.ErrQueueShutdown
 	}
-	return err
+	// send message
+	base := time.Now()
 
+	msg := kafkaAPI.Message{
+		Time:  base.Truncate(time.Millisecond),
+		Value: job.Bytes(),
+	}
+	if w.opts.compression == nil {
+		_, err = w.conn.WriteMessages(msg)
+	} else {
+		_, err = w.conn.WriteCompressedMessages(w.opts.compression, msg)
+	}
+	// if err != nil {
+	// 	t.Fatal(err)
+	// }
+	return err
 }
 
 func (w *Worker) Request() (core.QueuedMessage, error) {

options.go

@@ -5,19 +5,21 @@ import (
 
 	"github.com/golang-queue/queue"
 	"github.com/golang-queue/queue/core"
+	"github.com/segmentio/kafka-go/compress"
 )
 
 // Option for queue system
 type Option func(*options)
 
 type options struct {
-	runFunc   func(context.Context, core.QueuedMessage) error
-	logger    queue.Logger
-	addr      string
-	network   string
-	queue     string
-	topic     string
-	partition int //kafka's partition
+	runFunc     func(context.Context, core.QueuedMessage) error
+	logger      queue.Logger
+	addr        string
+	network     string
+	queue       string
+	topic       string
+	partition   int //kafka's partition
+	compression compress.Codec
 }
 
 // WithAddr setup the URI
@@ -54,6 +56,13 @@ func WithQueue(val string) Option {
 	}
 }
 
+// WithAddr setup the URI
+func WithCompress(compress compress.Codec) Option {
+	return func(w *options) {
+		w.compression = compress
+	}
+}
+
 // WithRunFunc setup the run func of queue
 func WithRunFunc(fn func(context.Context, core.QueuedMessage) error) Option {
 	return func(w *options) {