Merge pull request #71 from RealFax/dev

Dev

internal/rqd/raft_log_apply_test.go

@@ -1,19 +1,9 @@
 package rqd_test
 
 import (
-	"bytes"
-	"context"
+	"github.com/RealFax/RedQueen/api/serverpb"
 	red "github.com/RealFax/RedQueen/internal/rqd"
-	"github.com/RealFax/RedQueen/pkg/collapsar"
-	"io"
-	"strconv"
 	"testing"
-	"time"
-
-	"github.com/hashicorp/raft"
-	"google.golang.org/protobuf/proto"
-
-	"github.com/RealFax/RedQueen/api/serverpb"
 )
 
 var raftLogPayloadMessage = &serverpb.RaftLogPayload{
@@ -30,56 +20,6 @@ func TestRaftLogPayloadKey(t *testing.T) {
 	t.Logf("Payload key: %d", red.RaftLogPayloadKey(raftLogPayloadMessage))
 }
 
-func TestRaftMultipleLogApply_Apply(t *testing.T) {
-	rm := red.NewRaftMultipleLogApply(
-		context.Background(),
-		3,
-		time.Millisecond*300,
-		time.Millisecond*300,
-		func(cmd []byte, timeout time.Duration) raft.ApplyFuture {
-			r, err := collapsar.NewReader(bytes.NewReader(cmd[4:]))
-			if err != nil {
-				t.Fatal(err)
-			}
-
-			for {
-				pack, rErr := r.Next()
-				if rErr != nil {
-					if rErr == io.EOF {
-						break
-					}
-					t.Fatal(err)
-				}
-				m := &serverpb.RaftLogPayload{}
-				if err = proto.Unmarshal(pack, m); err != nil {
-					t.Fatal(err)
-				}
-				t.Log(m.String())
-			}
-			return nil
-		})
-
-	for i := 0; i < 3; i++ {
-		ctx := context.Background()
-		if err := rm.Apply(&ctx, raftLogPayloadMessage, 0); err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	time.Sleep(time.Second * 1)
-
-	for i := 0; i < 3; i++ {
-		ctx := context.Background()
-		raftLogPayloadMessage.Key = []byte("test_key" + strconv.Itoa(i))
-		if err := rm.Apply(&ctx, raftLogPayloadMessage, 0); err != nil {
-			t.Fatal(err)
-		}
-	}
-
-	time.Sleep(time.Second * 1)
-
-}
-
 func BenchmarkRaftLogPayloadKey(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		red.RaftLogPayloadKey(raftLogPayloadMessage)

pkg/collapsar/writer.go

@@ -21,7 +21,10 @@ func (a *writer) onFullTrigger() {
 	if atomic.LoadUint32(&a.state) == 0 || atomic.LoadInt32(&a.max) != atomic.LoadInt32(&a.size) {
 		return
 	}
-	_ = a.Close()
+
+	a.mu.Lock()
+	close(a.onFull)
+	a.mu.Unlock()
 }
 
 func (a *writer) Encode(w io.Writer) error {