From 600d2f92a44b8fb76653ef0547bce5a406e82eb8 Mon Sep 17 00:00:00 2001
From: chrislu <chris.lu@gmail.com>
Date: Sun, 25 Sep 2022 11:19:05 -0700
Subject: add message pipeline

---
 weed/mq/messages/message_pipeline_test.go | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)
 create mode 100644 weed/mq/messages/message_pipeline_test.go

(limited to 'weed/mq/messages/message_pipeline_test.go')

diff --git a/weed/mq/messages/message_pipeline_test.go b/weed/mq/messages/message_pipeline_test.go
new file mode 100644
index 000000000..aaa25e0fd
--- /dev/null
+++ b/weed/mq/messages/message_pipeline_test.go
@@ -0,0 +1,29 @@
+package messages
+
+import (
+	"testing"
+	"time"
+)
+
+func TestAddMessage(t *testing.T) {
+	mp := NewMessagePipeline(0, 3, 10, time.Second, &EmptyMover{})
+	go func() {
+		outChan := mp.OutputChan()
+		for mr := range outChan {
+			println(mr.sequence, mr.fileId)
+		}
+	}()
+
+	for i := 0; i < 100; i++ {
+		message := &Message{
+			Key:        []byte("key"),
+			Content:    []byte("data"),
+			Properties: nil,
+			Ts:         time.Now(),
+		}
+		mp.AddMessage(message)
+	}
+
+	mp.ShutdownStart()
+	mp.ShutdownWait()
+}
-- 
cgit v1.2.3