rangedbws

// Given
rangedbtest.SetRand(100)
inMemoryStore := inmemorystore.New(
	inmemorystore.WithClock(sequentialclock.New()),
	inmemorystore.WithUUIDGenerator(rangedbtest.NewSeededUUIDGenerator()),
)
api, err := rangedbws.New(rangedbws.WithStore(inMemoryStore))
PrintError(err)
defer api.Stop()

server := httptest.NewServer(api)
defer server.Close()

serverAddress := strings.TrimPrefix(server.URL, "http://")
websocketUrl := fmt.Sprintf("ws://%s/events/thing,that", serverAddress)
socket, _, err := websocket.DefaultDialer.Dial(websocketUrl, nil)
PrintError(err)

var wg sync.WaitGroup
wg.Add(1)

go func() {
	defer Close(socket)

	for i := 0; i < 2; i++ {
		_, message, err := socket.ReadMessage()
		PrintError(err)
		fmt.Println(jsontools.PrettyJSON(message))
	}

	wg.Done()
}()

// When
ctx, done := context.WithTimeout(context.Background(), 5*time.Second)
defer done()
PrintError(IgnoreFirstNumber(inMemoryStore.Save(ctx,
	&rangedb.EventRecord{Event: rangedbtest.ThingWasDone{ID: "dce275e43137467b92c9f4eb6c9c77a3", Number: 100}},
)))
PrintError(IgnoreFirstNumber(inMemoryStore.Save(ctx,
	&rangedb.EventRecord{Event: rangedbtest.AnotherWasComplete{ID: "594c68cfa7944f9b94afc83505ff99e9"}},
)))
PrintError(IgnoreFirstNumber(inMemoryStore.Save(ctx,
	&rangedb.EventRecord{Event: rangedbtest.ThatWasDone{ID: "075d37ae85894093aa818b391442df9b"}},
)))

wg.Wait()

Output:
{
  "aggregateType": "thing",
  "aggregateID": "dce275e43137467b92c9f4eb6c9c77a3",
  "globalSequenceNumber": 1,
  "streamSequenceNumber": 1,
  "insertTimestamp": 0,
  "eventID": "d2ba8e70072943388203c438d4e94bf3",
  "eventType": "ThingWasDone",
  "data": {
    "id": "dce275e43137467b92c9f4eb6c9c77a3",
    "number": 100
  },
  "metadata": null
}
{
  "aggregateType": "that",
  "aggregateID": "075d37ae85894093aa818b391442df9b",
  "globalSequenceNumber": 3,
  "streamSequenceNumber": 1,
  "insertTimestamp": 2,
  "eventID": "2e9e6918af10498cb7349c89a351fdb7",
  "eventType": "ThatWasDone",
  "data": {
    "ID": "075d37ae85894093aa818b391442df9b"
  },
  "metadata": null
}

// Given
rangedbtest.SetRand(100)
inMemoryStore := inmemorystore.New(
	inmemorystore.WithClock(sequentialclock.New()),
	inmemorystore.WithUUIDGenerator(rangedbtest.NewSeededUUIDGenerator()),
)
websocketApi, err := rangedbws.New(rangedbws.WithStore(inMemoryStore))
PrintError(err)
defer websocketApi.Stop()

server := httptest.NewServer(websocketApi)
defer server.Close()

serverAddress := strings.TrimPrefix(server.URL, "http://")
websocketUrl := fmt.Sprintf("ws://%s/events", serverAddress)
socket, _, err := websocket.DefaultDialer.Dial(websocketUrl, nil)
PrintError(err)

var wg sync.WaitGroup
wg.Add(1)

go func() {
	defer Close(socket)

	for i := 0; i < 2; i++ {
		_, message, err := socket.ReadMessage()
		PrintError(err)
		fmt.Println(jsontools.PrettyJSON(message))
	}

	wg.Done()
}()

// When
ctx, done := context.WithTimeout(context.Background(), 5*time.Second)
defer done()
PrintError(IgnoreFirstNumber(inMemoryStore.Save(ctx,
	&rangedb.EventRecord{Event: rangedbtest.ThingWasDone{ID: "52e247a7c0a54a65906e006dac9be108", Number: 100}},
)))
PrintError(IgnoreFirstNumber(inMemoryStore.Save(ctx,
	&rangedb.EventRecord{Event: rangedbtest.AnotherWasComplete{ID: "a3d9faa7614a46b388c6dce9984b6620"}},
)))

wg.Wait()

Output:
{
  "aggregateType": "thing",
  "aggregateID": "52e247a7c0a54a65906e006dac9be108",
  "globalSequenceNumber": 1,
  "streamSequenceNumber": 1,
  "insertTimestamp": 0,
  "eventID": "d2ba8e70072943388203c438d4e94bf3",
  "eventType": "ThingWasDone",
  "data": {
    "id": "52e247a7c0a54a65906e006dac9be108",
    "number": 100
  },
  "metadata": null
}
{
  "aggregateType": "another",
  "aggregateID": "a3d9faa7614a46b388c6dce9984b6620",
  "globalSequenceNumber": 2,
  "streamSequenceNumber": 1,
  "insertTimestamp": 1,
  "eventID": "99cbd88bbcaf482ba1cc96ed12541707",
  "eventType": "AnotherWasComplete",
  "data": {
    "id": "a3d9faa7614a46b388c6dce9984b6620"
  },
  "metadata": null
}