<commit_before>package store

import (
	"github.com/smancke/guble/protocol"

	"github.com/stretchr/testify/assert"

	"io/ioutil"
	"math"
	"os"
	"strconv"
	"testing"
	"time"
)

func Test_MessagePartition_foConcurrentWriteAndReads(t *testing.T) {
	//defer enableDebugForMethod()()
	a := assert.New(t)
	dir, _ := ioutil.TempDir("", "partition_store_test")
	defer os.RemoveAll(dir)

	store, _ := NewMessagePartition(dir, "myMessages")

	n := 1000 * 100

	writerDone := make(chan bool)
	readerDone := make(chan bool)
	go messagePartitionWriter(a, store, n, writerDone)
	go messagePartitionReader("reader1", a, store, n, readerDone)
	go messagePartitionReader("reader2", a, store, n, readerDone)
	go messagePartitionReader("reader3", a, store, n, readerDone)
	go messagePartitionReader("reader4", a, store, n, readerDone)
	go messagePartitionReader("reader5", a, store, n, readerDone)

	select {
	case <-writerDone:
	case <-time.After(time.Second * 15):
		a.Fail("writer timed out")
	}

	timeout := time.After(time.Second * 15)
	for i := 0; i < 5; i++ {
		select {
		case <-readerDone:
		case <-timeout:
			a.Fail("reader timed out")
		}
	}
}

func messagePartitionWriter(a *assert.Assertions, store *MessagePartition, n int, done chan bool) {
	for i := 1; i <= n; i++ {
		msg := []byte("Hello " + strconv.Itoa(i))
		a.NoError(store.Store(uint64(i), msg))
	}
	done <- true
}

func messagePartitionReader(name string, a *assert.Assertions, store *MessagePartition, n int, done chan bool) {
	lastReadMessage := 0
	for lastReadMessage < n {

		msgC := make(chan MessageAndId)
		errorC := make(chan error)

		protocol.Debug("[%v] start fetching at: %v", name, lastReadMessage+1)
		store.Fetch(FetchRequest{
			Partition:     "myMessages",
			StartId:       uint64(lastReadMessage + 1),
			Direction:     1,
			Count:         math.MaxInt32,
			MessageC:      msgC,
			ErrorCallback: errorC,
			StartCallback: make(chan int, 1),
		})

	fetch:

		for {
			select {
			case msgAndId, open := <-msgC:
				if !open {
					protocol.Debug("[%v] stop fetching at %v", name, lastReadMessage)
					break fetch
				}
				a.Equal(lastReadMessage+1, int(msgAndId.Id))
				lastReadMessage = int(msgAndId.Id)
			case err := <-errorC:
				a.Fail("received error", err.Error())
				<-done
				return
			}
		}

	}
	protocol.Debug("[%v] ready, got %v", name, lastReadMessage)
	done <- true
}
<commit_msg>refactoring test<commit_after>package store

import (
	"github.com/smancke/guble/protocol"

	"github.com/stretchr/testify/assert"

	"io/ioutil"
	"math"
	"os"
	"strconv"
	"testing"
	"time"
)

func Test_MessagePartition_forConcurrentWriteAndReads(t *testing.T) {
	//defer enableDebugForMethod()()
	a := assert.New(t)
	dir, _ := ioutil.TempDir("", "partition_store_test")
	defer os.RemoveAll(dir)

	store, _ := NewMessagePartition(dir, "myMessages")

	n := 1000 * 100
	nReaders := 5

	writerDone := make(chan bool)
	go messagePartitionWriter(a, store, n, writerDone)

	readerDone := make(chan bool)
	for i := 0; i < nReaders; i++ {
		go messagePartitionReader("reader" + strconv.Itoa(i), a, store, n, readerDone)
	}

	select {
	case <-writerDone:
	case <-time.After(time.Second * 15):
		a.Fail("writer timed out")
	}

	timeout := time.After(time.Second * 15)
	for i := 0; i < nReaders; i++ {
		select {
		case <-readerDone:
		case <-timeout:
			a.Fail("reader timed out")
		}
	}
}

func messagePartitionWriter(a *assert.Assertions, store *MessagePartition, n int, done chan bool) {
	for i := 1; i <= n; i++ {
		msg := []byte("Hello " + strconv.Itoa(i))
		a.NoError(store.Store(uint64(i), msg))
	}
	done <- true
}

func messagePartitionReader(name string, a *assert.Assertions, store *MessagePartition, n int, done chan bool) {
	lastReadMessage := 0
	for lastReadMessage < n {

		msgC := make(chan MessageAndId)
		errorC := make(chan error)

		protocol.Debug("[%v] start fetching at: %v", name, lastReadMessage+1)
		store.Fetch(FetchRequest{
			Partition:     "myMessages",
			StartId:       uint64(lastReadMessage + 1),
			Direction:     1,
			Count:         math.MaxInt32,
			MessageC:      msgC,
			ErrorCallback: errorC,
			StartCallback: make(chan int, 1),
		})

	fetch:

		for {
			select {
			case msgAndId, open := <-msgC:
				if !open {
					protocol.Debug("[%v] stop fetching at %v", name, lastReadMessage)
					break fetch
				}
				a.Equal(lastReadMessage+1, int(msgAndId.Id))
				lastReadMessage = int(msgAndId.Id)
			case err := <-errorC:
				a.Fail("received error", err.Error())
				<-done
				return
			}
		}

	}
	protocol.Debug("[%v] ready, got %v", name, lastReadMessage)
	done <- true
}

{'fraction_non_alphanumeric': 0.1479134804457068, 'fraction_numerical': 0.01267205593183308, 'mean_word_length': 3.7244582043343653, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7578248', 'n_tokens_mistral': 1868, 'n_tokens_neox': 1644, 'n_words': 396}