mirror of
https://github.com/gogrlx/nats-server.git
synced 2026-04-02 11:48:43 -07:00
244 lines
7.1 KiB
Go
244 lines
7.1 KiB
Go
// Copyright 2019-2020 The NATS Authors
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package server
|
|
|
|
import (
|
|
"bytes"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestMemStoreBasics(t *testing.T) {
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
|
|
subj, msg := "foo", []byte("Hello World")
|
|
now := time.Now().UnixNano()
|
|
if seq, ts, err := ms.StoreMsg(subj, nil, msg); err != nil {
|
|
t.Fatalf("Error storing msg: %v", err)
|
|
} else if seq != 1 {
|
|
t.Fatalf("Expected sequence to be 1, got %d", seq)
|
|
} else if ts < now || ts > now+int64(time.Millisecond) {
|
|
t.Fatalf("Expected timestamp to be current, got %v", ts-now)
|
|
}
|
|
|
|
state := ms.State()
|
|
if state.Msgs != 1 {
|
|
t.Fatalf("Expected 1 msg, got %d", state.Msgs)
|
|
}
|
|
expectedSize := memStoreMsgSize(subj, nil, msg)
|
|
if state.Bytes != expectedSize {
|
|
t.Fatalf("Expected %d bytes, got %d", expectedSize, state.Bytes)
|
|
}
|
|
nsubj, _, nmsg, _, err := ms.LoadMsg(1)
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error looking up msg: %v", err)
|
|
}
|
|
if nsubj != subj {
|
|
t.Fatalf("Subjects don't match, original %q vs %q", subj, nsubj)
|
|
}
|
|
if !bytes.Equal(nmsg, msg) {
|
|
t.Fatalf("Msgs don't match, original %q vs %q", msg, nmsg)
|
|
}
|
|
}
|
|
|
|
func TestMemStoreMsgLimit(t *testing.T) {
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage, MaxMsgs: 10})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
subj, msg := "foo", []byte("Hello World")
|
|
for i := 0; i < 10; i++ {
|
|
ms.StoreMsg(subj, nil, msg)
|
|
}
|
|
state := ms.State()
|
|
if state.Msgs != 10 {
|
|
t.Fatalf("Expected %d msgs, got %d", 10, state.Msgs)
|
|
}
|
|
if _, _, err := ms.StoreMsg(subj, nil, msg); err != nil {
|
|
t.Fatalf("Error storing msg: %v", err)
|
|
}
|
|
state = ms.State()
|
|
if state.Msgs != 10 {
|
|
t.Fatalf("Expected %d msgs, got %d", 10, state.Msgs)
|
|
}
|
|
if state.LastSeq != 11 {
|
|
t.Fatalf("Expected the last sequence to be 11 now, but got %d", state.LastSeq)
|
|
}
|
|
if state.FirstSeq != 2 {
|
|
t.Fatalf("Expected the first sequence to be 2 now, but got %d", state.FirstSeq)
|
|
}
|
|
// Make sure we can not lookup seq 1.
|
|
if _, _, _, _, err := ms.LoadMsg(1); err == nil {
|
|
t.Fatalf("Expected error looking up seq 1 but got none")
|
|
}
|
|
}
|
|
|
|
func TestMemStoreBytesLimit(t *testing.T) {
|
|
subj, msg := "foo", make([]byte, 512)
|
|
storedMsgSize := memStoreMsgSize(subj, nil, msg)
|
|
|
|
toStore := uint64(1024)
|
|
maxBytes := storedMsgSize * toStore
|
|
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage, MaxBytes: int64(maxBytes)})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
|
|
for i := uint64(0); i < toStore; i++ {
|
|
ms.StoreMsg(subj, nil, msg)
|
|
}
|
|
state := ms.State()
|
|
if state.Msgs != toStore {
|
|
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
|
|
}
|
|
if state.Bytes != storedMsgSize*toStore {
|
|
t.Fatalf("Expected bytes to be %d, got %d", storedMsgSize*toStore, state.Bytes)
|
|
}
|
|
|
|
// Now send 10 more and check that bytes limit enforced.
|
|
for i := 0; i < 10; i++ {
|
|
if _, _, err := ms.StoreMsg(subj, nil, msg); err != nil {
|
|
t.Fatalf("Error storing msg: %v", err)
|
|
}
|
|
}
|
|
state = ms.State()
|
|
if state.Msgs != toStore {
|
|
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
|
|
}
|
|
if state.Bytes != storedMsgSize*toStore {
|
|
t.Fatalf("Expected bytes to be %d, got %d", storedMsgSize*toStore, state.Bytes)
|
|
}
|
|
if state.FirstSeq != 11 {
|
|
t.Fatalf("Expected first sequence to be 11, got %d", state.FirstSeq)
|
|
}
|
|
if state.LastSeq != toStore+10 {
|
|
t.Fatalf("Expected last sequence to be %d, got %d", toStore+10, state.LastSeq)
|
|
}
|
|
}
|
|
|
|
func TestMemStoreAgeLimit(t *testing.T) {
|
|
maxAge := 10 * time.Millisecond
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage, MaxAge: maxAge})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
// Store some messages. Does not really matter how many.
|
|
subj, msg := "foo", []byte("Hello World")
|
|
toStore := 100
|
|
for i := 0; i < toStore; i++ {
|
|
ms.StoreMsg(subj, nil, msg)
|
|
}
|
|
state := ms.State()
|
|
if state.Msgs != uint64(toStore) {
|
|
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
|
|
}
|
|
checkExpired := func(t *testing.T) {
|
|
t.Helper()
|
|
checkFor(t, time.Second, maxAge, func() error {
|
|
state = ms.State()
|
|
if state.Msgs != 0 {
|
|
return fmt.Errorf("Expected no msgs, got %d", state.Msgs)
|
|
}
|
|
if state.Bytes != 0 {
|
|
return fmt.Errorf("Expected no bytes, got %d", state.Bytes)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
// Let them expire
|
|
checkExpired(t)
|
|
// Now add some more and make sure that timer will fire again.
|
|
for i := 0; i < toStore; i++ {
|
|
ms.StoreMsg(subj, nil, msg)
|
|
}
|
|
state = ms.State()
|
|
if state.Msgs != uint64(toStore) {
|
|
t.Fatalf("Expected %d msgs, got %d", toStore, state.Msgs)
|
|
}
|
|
checkExpired(t)
|
|
}
|
|
|
|
func TestMemStoreTimeStamps(t *testing.T) {
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
last := time.Now().UnixNano()
|
|
subj, msg := "foo", []byte("Hello World")
|
|
for i := 0; i < 10; i++ {
|
|
time.Sleep(5 * time.Microsecond)
|
|
ms.StoreMsg(subj, nil, msg)
|
|
}
|
|
for seq := uint64(1); seq <= 10; seq++ {
|
|
_, _, _, ts, err := ms.LoadMsg(seq)
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error looking up msg: %v", err)
|
|
}
|
|
// These should be different
|
|
if ts <= last {
|
|
t.Fatalf("Expected different timestamps, got %v", ts)
|
|
}
|
|
last = ts
|
|
}
|
|
}
|
|
|
|
func TestMemStoreEraseMsg(t *testing.T) {
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
subj, msg := "foo", []byte("Hello World")
|
|
ms.StoreMsg(subj, nil, msg)
|
|
_, _, smsg, _, err := ms.LoadMsg(1)
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error looking up msg: %v", err)
|
|
}
|
|
if !bytes.Equal(msg, smsg) {
|
|
t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
|
|
}
|
|
if removed, _ := ms.EraseMsg(1); !removed {
|
|
t.Fatalf("Expected erase msg to return success")
|
|
}
|
|
}
|
|
|
|
func TestMemStoreMsgHeaders(t *testing.T) {
|
|
ms, err := newMemStore(&StreamConfig{Storage: MemoryStorage})
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error creating store: %v", err)
|
|
}
|
|
subj, hdr, msg := "foo", []byte("name:derek"), []byte("Hello World")
|
|
if sz := int(memStoreMsgSize(subj, hdr, msg)); sz != (len(subj) + len(hdr) + len(msg) + 16) {
|
|
t.Fatalf("Wrong size for stored msg with header")
|
|
}
|
|
ms.StoreMsg(subj, hdr, msg)
|
|
_, shdr, smsg, _, err := ms.LoadMsg(1)
|
|
if err != nil {
|
|
t.Fatalf("Unexpected error looking up msg: %v", err)
|
|
}
|
|
if !bytes.Equal(msg, smsg) {
|
|
t.Fatalf("Expected same msg, got %q vs %q", smsg, msg)
|
|
}
|
|
if !bytes.Equal(hdr, shdr) {
|
|
t.Fatalf("Expected same hdr, got %q vs %q", shdr, hdr)
|
|
}
|
|
if removed, _ := ms.EraseMsg(1); !removed {
|
|
t.Fatalf("Expected erase msg to return success")
|
|
}
|
|
}
|