132 lines
3.0 KiB
Go
132 lines
3.0 KiB
Go
package nats
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// TestNatsBasicOperations 测试基础操作
|
|
func TestNatsBasicOperations(t *testing.T) {
|
|
// 测试连接状态
|
|
if !IsConnected() {
|
|
t.Log("NATS 未连接")
|
|
}
|
|
|
|
// 测试连接状态获取
|
|
state := GetConnState()
|
|
t.Logf("当前连接状态: %d", state)
|
|
}
|
|
|
|
// TestNatsMetrics 测试监控指标
|
|
func TestNatsMetrics(t *testing.T) {
|
|
metrics := GetMetrics()
|
|
t.Logf("发布计数: %d", metrics.PublishCount.Load())
|
|
t.Logf("发布错误: %d", metrics.PublishError.Load())
|
|
t.Logf("请求计数: %d", metrics.RequestCount.Load())
|
|
}
|
|
|
|
// TestNatsConnStateListener 测试连接状态监听
|
|
func TestNatsConnStateListener(t *testing.T) {
|
|
listener := func(state ConnState, err error) {
|
|
fmt.Printf("连接状态变化: %d, 错误: %v\n", state, err)
|
|
}
|
|
|
|
RegisterConnStateListener(listener)
|
|
defer UnregisterConnStateListener(listener)
|
|
|
|
time.Sleep(1 * time.Second)
|
|
}
|
|
|
|
// TestNatsStreamOperations 测试流操作
|
|
func TestNatsStreamOperations(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
// 创建任务流
|
|
config := TaskStreamConfig{
|
|
StreamName: "test_tasks",
|
|
Subjects: []string{"test.task.>"},
|
|
//Subject: "test.task.process",
|
|
}
|
|
err := CreateTaskStream(ctx, config)
|
|
if err != nil {
|
|
t.Logf("创建任务流失败: %v", err)
|
|
}
|
|
|
|
// 获取流信息
|
|
info, err := GetStream(ctx, "test_tasks")
|
|
if err != nil {
|
|
t.Logf("获取流信息失败: %v", err)
|
|
} else {
|
|
t.Logf("流信息: %s", info.Config.Name)
|
|
}
|
|
|
|
// 列出所有流
|
|
streams, err := ListStreams(ctx)
|
|
if err != nil {
|
|
t.Logf("列出流失败: %v", err)
|
|
} else {
|
|
t.Logf("流列表: %v", streams)
|
|
}
|
|
|
|
// 删除流
|
|
err = DeleteStream(ctx, "test_tasks")
|
|
if err != nil {
|
|
t.Logf("删除流失败: %v", err)
|
|
}
|
|
}
|
|
|
|
// TestNatsConsumerOperations 测试消费者操作
|
|
func TestNatsConsumerOperations(t *testing.T) {
|
|
ctx := context.Background()
|
|
|
|
// 创建测试流
|
|
config := TaskStreamConfig{
|
|
StreamName: "test_consumer",
|
|
Subjects: []string{"test.consumer.>"},
|
|
//Subject: "test.consumer.process",
|
|
}
|
|
err := CreateTaskStream(ctx, config)
|
|
if err != nil {
|
|
t.Logf("创建流失败: %v", err)
|
|
}
|
|
|
|
// 创建消费者
|
|
consumerConfig := jetstream.ConsumerConfig{
|
|
Name: "test_consumer",
|
|
Durable: "test_consumer",
|
|
}
|
|
_, err = CreateConsumer(ctx, "test_consumer", "test_consumer", consumerConfig)
|
|
if err != nil {
|
|
t.Logf("创建消费者失败: %v", err)
|
|
}
|
|
|
|
// 获取消费者信息
|
|
info, err := GetConsumer(ctx, "test_consumer", "test_consumer")
|
|
if err != nil {
|
|
t.Logf("获取消费者信息失败: %v", err)
|
|
} else {
|
|
t.Logf("消费者信息: %s", info.Name)
|
|
}
|
|
|
|
// 列出消费者
|
|
consumers, err := ListConsumers(ctx, "test_consumer")
|
|
if err != nil {
|
|
t.Logf("列出消费者失败: %v", err)
|
|
} else {
|
|
t.Logf("消费者列表: %v", consumers)
|
|
}
|
|
|
|
// 删除消费者
|
|
err = DeleteConsumer(ctx, "test_consumer", "test_consumer")
|
|
if err != nil {
|
|
t.Logf("删除消费者失败: %v", err)
|
|
}
|
|
|
|
// 清理流
|
|
_ = DeleteStream(ctx, "test_consumer")
|
|
}
|