<commit_before>package coal

import (
	"testing"
	"time"

	"github.com/globalsign/mgo/bson"
	"github.com/stretchr/testify/assert"
)

func TestStream(t *testing.T) {
	tester.Clean()

	time.Sleep(100 * time.Millisecond)

	i := 1
	open := make(chan struct{})
	done := make(chan struct{})

	stream := OpenStream(tester.Store, &postModel{}, nil, func(e Event, id bson.ObjectId, m Model, token []byte) {
		switch i {
		case 1:
			assert.Equal(t, Created, e)
			assert.NotZero(t, id)
			assert.NotNil(t, m)
			assert.NotNil(t, token)
		case 2:
			assert.Equal(t, Updated, e)
			assert.NotZero(t, id)
			assert.NotNil(t, m)
			assert.NotNil(t, token)
		case 3:
			assert.Equal(t, Deleted, e)
			assert.NotZero(t, id)
			assert.Nil(t, m)
			assert.NotNil(t, token)

			close(done)
		}

		i++
	}, func() {
		close(open)
	}, func(err error) {
		panic(err)
	})

	<-open

	s := tester.Store.Copy()
	defer s.Close()

	post := Init(&postModel{
		Title: "foo",
	}).(*postModel)

	err := s.C(post).Insert(post)
	assert.NoError(t, err)

	post.Title = "bar"

	err = s.C(post).UpdateId(post.ID(), post)
	assert.NoError(t, err)

	err = s.C(post).RemoveId(post.ID())
	assert.NoError(t, err)

	<-done

	stream.Close()
}<commit_msg>test stream resumption<commit_after>package coal

import (
	"testing"
	"time"

	"github.com/globalsign/mgo/bson"
	"github.com/stretchr/testify/assert"
)

func TestStream(t *testing.T) {
	tester.Clean()

	time.Sleep(100 * time.Millisecond)

	i := 1
	open := make(chan struct{})
	done := make(chan struct{})

	stream := OpenStream(tester.Store, &postModel{}, nil, func(e Event, id bson.ObjectId, m Model, token []byte) {
		switch i {
		case 1:
			assert.Equal(t, Created, e)
			assert.NotZero(t, id)
			assert.NotNil(t, m)
			assert.NotNil(t, token)
		case 2:
			assert.Equal(t, Updated, e)
			assert.NotZero(t, id)
			assert.NotNil(t, m)
			assert.NotNil(t, token)
		case 3:
			assert.Equal(t, Deleted, e)
			assert.NotZero(t, id)
			assert.Nil(t, m)
			assert.NotNil(t, token)

			close(done)
		}

		i++
	}, func() {
		close(open)
	}, func(err error) {
		panic(err)
	})

	<-open

	s := tester.Store.Copy()
	defer s.Close()

	post := Init(&postModel{
		Title: "foo",
	}).(*postModel)

	err := s.C(post).Insert(post)
	assert.NoError(t, err)

	post.Title = "bar"

	err = s.C(post).UpdateId(post.ID(), post)
	assert.NoError(t, err)

	err = s.C(post).RemoveId(post.ID())
	assert.NoError(t, err)

	<-done

	stream.Close()
}

func TestStreamResumption(t *testing.T) {
	tester.Clean()

	time.Sleep(100 * time.Millisecond)

	open1 := make(chan struct{})
	done1 := make(chan struct{})

	var resumeToken []byte

	stream1 := OpenStream(tester.Store, &postModel{}, nil, func(e Event, id bson.ObjectId, m Model, token []byte) {
		assert.Equal(t, Created, e)
		assert.NotZero(t, id)
		assert.NotNil(t, m)
		assert.NotNil(t, token)

		resumeToken = token
		close(done1)
	}, func() {
		close(open1)
	}, func(err error) {
		panic(err)
	})

	<-open1

	s := tester.Store.Copy()
	defer s.Close()

	post := Init(&postModel{
		Title: "foo",
	}).(*postModel)

	err := s.C(post).Insert(post)
	assert.NoError(t, err)

	<-done1
	stream1.Close()

	post.Title = "bar"

	err = s.C(post).UpdateId(post.ID(), post)
	assert.NoError(t, err)

	err = s.C(post).RemoveId(post.ID())
	assert.NoError(t, err)

	i := 1
	done2 := make(chan struct{})

	stream2 := OpenStream(tester.Store, &postModel{}, resumeToken, func(e Event, id bson.ObjectId, m Model, token []byte) {
		switch i {
		case 1:
			assert.Equal(t, Updated, e)
			assert.NotZero(t, id)
			assert.NotNil(t, m)
			assert.NotNil(t, token)
		case 2:
			assert.Equal(t, Deleted, e)
			assert.NotZero(t, id)
			assert.Nil(t, m)
			assert.NotNil(t, token)

			close(done2)
		}

		i++
	}, nil, func(err error) {
		panic(err)
	})

	<-done2

	stream2.Close()
}
{'fraction_non_alphanumeric': 0.18064004231684738, 'fraction_numerical': 0.008727849775191749, 'mean_word_length': 3.507747318235995, 'pattern_counts': {'":': 0, '<': 10, '<?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': '2533882', 'n_tokens_mistral': 1731, 'n_tokens_neox': 1548, 'n_words': 326}