{#if event}
-
- {#if event.type === MessageChannelEvent.MessageSent || event.type === MessageChannelEvent.MessageReceived}
+ {#if event.type === MessageChannelEvent.MessageSent || event.type === MessageChannelEvent.MessageReceived || event.type === MessageChannelEvent.SyncSent || event.type === MessageChannelEvent.SyncReceived}
{#each event.payload.causalHistory as dependency}
{@const dependencyMatchesFilter =
currentIdFilter && dependency.messageId === currentIdFilter}
@@ -81,6 +77,8 @@
padding: 8px;
box-sizing: border-box;
transition: transform 0.2s ease;
+ width: 100%;
+ max-width: 100%;
}
.history-item:not(.empty):hover {
@@ -104,6 +102,7 @@
gap: 6px;
width: 100%;
height: 100%;
+ overflow: hidden;
}
.event-box {
@@ -120,6 +119,8 @@
position: relative;
transition: all 0.2s ease;
border: none;
+ box-sizing: border-box;
+ overflow: hidden;
}
.dependency-box {
@@ -140,11 +141,12 @@
transition: all 0.2s ease;
overflow: hidden;
text-overflow: ellipsis;
+ box-sizing: border-box;
}
.highlight {
- border-left: 2px solid #DB8D43;
- border-right: 2px solid #DB8D43;
+ border-left: 2px solid #db8d43;
+ border-right: 2px solid #db8d43;
}
.highlight .event-type {
@@ -199,6 +201,7 @@
max-width: 220px;
text-overflow: ellipsis;
white-space: nowrap;
+ overflow: hidden;
}
.lamport-timestamp {
@@ -224,4 +227,4 @@
padding: 2px 6px;
border-radius: 4px;
}
-
\ No newline at end of file
+
diff --git a/examples/sds-demo/src/lib/components/Missing.svelte b/examples/sds-demo/src/lib/components/Missing.svelte
index f083aa7..991fe4d 100644
--- a/examples/sds-demo/src/lib/components/Missing.svelte
+++ b/examples/sds-demo/src/lib/components/Missing.svelte
@@ -1,43 +1,105 @@
+
@@ -48,18 +49,13 @@
{id}
- {#if event.type === MessageChannelEvent.MessageDelivered}
-
- {event.payload.sentOrReceived}
-
- {/if}
- {#if event.type === MessageChannelEvent.MessageSent || event.type === MessageChannelEvent.MessageReceived}
+ {#if event.type === MessageChannelEvent.MessageSent || event.type === MessageChannelEvent.MessageReceived || event.type === MessageChannelEvent.SyncSent || event.type === MessageChannelEvent.SyncReceived}
{/if}
- {#each missedMessages as entry, index}
- {@const color = eventColors[MessageChannelEvent.MissedMessages]}
- ());
-
+
// Method to update the state of a peer
updatePeerState(peerId: string, newState: PeerState, message: LobbyMessage): void {
const prevState = this.peerState.get(peerId)?.state || PeerState.None;
const messages = this.peerState.get(peerId)?.messages || [];
-
+
// Update the state
- this.peerState.set(peerId, {
- state: newState,
- messages: [...messages, message]
+ this.peerState.set(peerId, {
+ state: newState,
+ messages: [...messages, message]
});
-
+
// Create and dispatch the event using native event handling
const event = new CustomEvent('state-transition', {
detail: {
@@ -79,7 +84,7 @@ class LobbyState extends EventTarget {
message
}
});
-
+
this.dispatchEvent(event);
}
}
@@ -104,26 +109,29 @@ type TransitionTable = {
const stateMachine: TransitionTable = {
[LobbyEvent.GotPing]: {
[PeerState.None]: PeerState.Found,
- [PeerState.Found]: PeerState.Found,
+ [PeerState.Found]: PeerState.Found
},
[LobbyEvent.SentRequest]: {
- [PeerState.Found]: PeerState.RequestTo,
+ [PeerState.Found]: PeerState.RequestTo
},
[LobbyEvent.GotRequest]: {
[PeerState.None]: PeerState.RequestFrom,
- [PeerState.Found]: PeerState.RequestFrom,
+ [PeerState.Found]: PeerState.RequestFrom
},
[LobbyEvent.SentAccept]: {
- [PeerState.RequestFrom]: PeerState.AcceptTo,
+ [PeerState.RequestFrom]: PeerState.AcceptTo
},
[LobbyEvent.GotAccept]: {
- [PeerState.RequestTo]: PeerState.AcceptFrom,
+ [PeerState.RequestTo]: PeerState.AcceptFrom
},
[LobbyEvent.SentMatch]: {
- [PeerState.AcceptFrom]: PeerState.Success,
+ [PeerState.AcceptFrom]: PeerState.Success
},
[LobbyEvent.GotMatch]: {
- [PeerState.AcceptTo]: PeerState.Success,
+ [PeerState.AcceptTo]: PeerState.Success
+ },
+ [LobbyEvent.Ongoing]: {
+ [PeerState.None]: PeerState.Ongoing
}
};
@@ -142,19 +150,18 @@ function processMessage(
throw `Don't track sent pings`;
}
event = LobbyEvent.GotPing;
- } else if (
- message.messageType === LobbyMessageType.Request
- ) {
+ } else if (message.messageType === LobbyMessageType.Request) {
console.log(`Received request from ${message.from} to ${message.to || 'everyone'}`);
event = sent ? LobbyEvent.SentRequest : LobbyEvent.GotRequest;
- } else if (
- message.messageType === LobbyMessageType.Accept
- ) {
+ } else if (message.messageType === LobbyMessageType.Accept) {
console.log(`Received accept from ${message.from} to ${message.to || 'unknown'}`);
event = sent ? LobbyEvent.SentAccept : LobbyEvent.GotAccept;
} else if (message.messageType === LobbyMessageType.Match) {
console.log(`Received match between peers`);
event = sent ? LobbyEvent.SentMatch : LobbyEvent.GotMatch;
+ } else if (message.messageType === LobbyMessageType.Ongoing) {
+ console.log(`Received ongoing match`);
+ event = LobbyEvent.Ongoing;
}
// Get next state from transition table
@@ -162,7 +169,10 @@ function processMessage(
console.warn(`Invalid message type: ${message.messageType}`);
return Option.none();
}
- const nextStateValue = stateMachine[event][currentState];
+ const nextStateValue =
+ message.messageType === LobbyMessageType.Ongoing
+ ? PeerState.Ongoing
+ : stateMachine[event][currentState];
if (nextStateValue === undefined) {
// Handle invalid transitions - throw error or return current state
console.warn(`Invalid transition: ${event} from ${currentState}`);
@@ -171,7 +181,9 @@ function processMessage(
return Option.some(nextStateValue);
}
-export async function processUpdates(updates: { peerId: string; message: LobbyMessage, sent: boolean }[]) {
+export async function processUpdates(
+ updates: { peerId: string; message: LobbyMessage; sent: boolean }[]
+) {
for (const update of updates) {
const { peerId, message, sent } = update;
const currentState = lobbyState.peerState.get(peerId)?.state || PeerState.None;
@@ -189,28 +201,33 @@ export async function processUpdates(updates: { peerId: string; message: LobbyMe
}
// Create a typed stream from the events
-export const stateTransitionStream = $state(Stream.map(
- Stream.fromEventListener(lobbyState, 'state-transition', { passive: true }),
- (event: Event) => event as CustomEvent
-));
+export const stateTransitionStream = $state(
+ Stream.map(
+ Stream.fromEventListener(lobbyState, 'state-transition', { passive: true }),
+ (event: Event) => event as CustomEvent
+ )
+);
-export function subscribeToStateTransitionStream(stream: Stream.Stream>, onEvent: (event: A) => void): () => void {
+export function subscribeToStateTransitionStream(
+ stream: Stream.Stream>,
+ onEvent: (event: A) => void
+): () => void {
const fiber = Effect.runFork(
- pipe(
- stream,
- Stream.tap((event) =>
- Effect.sync(() => {
- onEvent(event.detail);
- })
- ),
- Stream.runDrain
- )
+ pipe(
+ stream,
+ Stream.tap((event) =>
+ Effect.sync(() => {
+ onEvent(event.detail);
+ })
+ ),
+ Stream.runDrain
+ )
);
return () => {
- Effect.runFork(
- Effect.sync(() => {
- (fiber as unknown as { interrupt: () => void }).interrupt();
- })
- );
+ Effect.runFork(
+ Effect.sync(() => {
+ (fiber as unknown as { interrupt: () => void }).interrupt();
+ })
+ );
};
- }
\ No newline at end of file
+}
diff --git a/examples/sds-demo/src/lib/waku/pingpong.svelte.ts b/examples/sds-demo/src/lib/waku/pingpong.svelte.ts
index ec52413..65c9429 100644
--- a/examples/sds-demo/src/lib/waku/pingpong.svelte.ts
+++ b/examples/sds-demo/src/lib/waku/pingpong.svelte.ts
@@ -1,14 +1,17 @@
-import { decoder, encoder, wakuNode } from '$lib/waku/waku.svelte';
+import { decoder, encoder, lobbyEncoder, wakuNode } from '$lib/waku/waku.svelte';
import { type Message, encodeMessage, decodeMessage } from '@waku/sds';
import type { MatchParams } from './waku.svelte';
import { getOrCreateChannel } from '$lib/sds/channel.svelte';
import { Effect, Schedule, Stream, Option, Queue, Chunk, Ref } from 'effect';
-import type { SubscribeResult } from '@waku/sdk';
+import type { SDKProtocolResult, SubscribeResult } from '@waku/sdk';
import { messageHash } from '@waku/message-hash';
import { hash } from '$lib/utils/hash';
import { encodeBase64 } from 'effect/Encoding';
import type { MessageChannel } from '@waku/sds';
import { sweepIn, sweepOut } from '$lib/sds.svelte';
+import { LobbyMessageType, type LobbyMessage } from './lobby.svelte';
+
+export const listening = $state({ listening: false });
export function send(channel: MessageChannel, payload: Uint8Array) {
return channel.sendMessage(payload, async (message: Message) => {
@@ -23,7 +26,6 @@ export function send(channel: MessageChannel, payload: Uint8Array) {
if (result.failures.length > 0) {
console.error('error sending message', result.failures);
}
- console.log('sent message over waku', message);
return {
success: true,
retrievalHint: hash
@@ -31,10 +33,30 @@ export function send(channel: MessageChannel, payload: Uint8Array) {
});
}
-export function start(params: MatchParams) {
- console.log('starting pingpong', params);
+const broadcastMatchToLobby = (params: MatchParams) =>
+ Effect.async((resume) => {
+ const m: LobbyMessage = {
+ messageType: LobbyMessageType.Ongoing,
+ timestamp: new Date(),
+ from: params.myPeerId,
+ to: params.otherPeerId,
+ match: params
+ };
+ wakuNode.node?.lightPush
+ .send(lobbyEncoder, {
+ payload: new TextEncoder().encode(JSON.stringify(m)),
+ timestamp: new Date()
+ })
+ .then((result) => resume(Effect.succeed(result)))
+ .catch((error) => resume(Effect.fail(new Error(error as string))));
+ });
+
+export function start(params: MatchParams, joined: boolean = false) {
const { matchId } = params;
- const first = params.myPeerId.localeCompare(params.otherPeerId) < 0 ? true : false;
+ const first =
+ params.myPeerId.localeCompare(params.otherPeerId + (joined ? Math.random() : '')) < 0
+ ? true
+ : false;
let lastMessage = new Date();
const sinkTimeout = 10_000;
@@ -44,7 +66,9 @@ export function start(params: MatchParams) {
// we can predetermine the list of messages we expect to send
const payloadsStream = Stream.make(
...Array.from({ length: params.messages }, (_, i) => {
- return new TextEncoder().encode(hash(matchId + params.myPeerId + i));
+ return new TextEncoder().encode(
+ hash(matchId + (joined ? Math.random() : params.myPeerId) + i)
+ );
})
);
const skips = Array.from({ length: params.messages * 2 }, (_, i) =>
@@ -85,7 +109,6 @@ export function start(params: MatchParams) {
// Initialize the queue with the payloads based on strategy
const initializeQueue = Effect.gen(function* () {
const sendQueue = Queue.bounded>(100);
- console.log('initializing queue');
const q = yield* sendQueue;
const result = yield* q.offerAll(yield* payloadsWithSkips);
console.log('queue initialized', result);
@@ -94,14 +117,11 @@ export function start(params: MatchParams) {
const takeAndSend = (sendQueue: Queue.Queue>) =>
Effect.gen(function* () {
- console.log('taking and sending');
const result = yield* sendQueue.take;
- console.log('result', result);
return yield* sendEffect(result);
});
const validateMessage = (message: Message) => {
- console.log('validating message', message);
if (message.channelId !== channel.channelId) {
console.error('Message is not for this match');
return false;
@@ -112,15 +132,12 @@ export function start(params: MatchParams) {
const subscribe = (listenCondition: Ref.Ref) =>
Effect.async((resume) => {
try {
- console.log('subscribing to filter');
wakuNode.node?.filter
.subscribe([decoder], async (message) => {
const listening = await Effect.runPromise(listenCondition.get);
- console.log('listening', listening);
if (!listening) {
return;
}
- console.log('received filter message', message);
const hash = encodeBase64(messageHash(encoder.pubsubTopic, message));
const sdsMessage = decodeMessage(message.payload) as unknown as Message;
if (validateMessage(sdsMessage) && !sent.has(hash)) {
@@ -140,9 +157,7 @@ export function start(params: MatchParams) {
const sent = new Map();
const sendSync = Effect.async((resume) => {
- console.log('sending sync', new Date().getTime() - lastMessage.getTime());
if (new Date().getTime() - lastMessage.getTime() < sinkTimeout + Math.random() * 2000) {
- console.log('sink timeout', new Date().getTime() - lastMessage.getTime());
return resume(Effect.succeed(false));
} else {
channel
@@ -151,7 +166,7 @@ export function start(params: MatchParams) {
const timestamp = new Date();
const protoMessage = await encoder.toProtoObj({
payload: encodedMessage,
- timestamp,
+ timestamp
});
const hash = encodeBase64(messageHash(encoder.pubsubTopic, protoMessage));
sent.set(hash, true);
@@ -159,7 +174,6 @@ export function start(params: MatchParams) {
if (result.failures.length > 0) {
console.error('error sending message', result.failures);
}
- console.log('sent message over waku', message);
lastMessage = new Date();
return true;
})
@@ -194,11 +208,18 @@ export function start(params: MatchParams) {
// periodically process queue
Effect.repeat(processQueue, Schedule.spaced('200 millis')),
// periodically sweep buffers
- Effect.repeat(sweep, Schedule.spaced('500 millis')),
+ Effect.repeat(sweep, Schedule.spaced('2000 millis')),
// periodically switch off filter to miss messages
Effect.repeat(
- Ref.update(listenCondition, (listening) => !listening),
- Schedule.spaced(first ? '5000 millis' : `${5000 * 2} millis`)
+ Ref.update(listenCondition, (_listening) => {
+ listening.listening = !_listening;
+ return !_listening;
+ }),
+ Schedule.spaced(first ? '4500 millis' : `8000 millis`)
+ ),
+ Effect.repeat(
+ broadcastMatchToLobby(params),
+ Schedule.spaced('2000 millis')
)
],
{
diff --git a/examples/sds-demo/src/lib/waku/waku.svelte.ts b/examples/sds-demo/src/lib/waku/waku.svelte.ts
index 8a65f12..2949e87 100644
--- a/examples/sds-demo/src/lib/waku/waku.svelte.ts
+++ b/examples/sds-demo/src/lib/waku/waku.svelte.ts
@@ -78,7 +78,7 @@ export class WakuNode {
}
return await node.lightPush.send(encoder, {
payload: payload,
- timestamp: timestamp,
+ timestamp: timestamp
});
}
@@ -133,10 +133,19 @@ export async function startWaku(): Promise {
await wakuNode.setNode(node);
// Connect to peers
- await node.dial(
- // '/dns4/waku-test.bloxy.one/tcp/8095/wss/p2p/16Uiu2HAmSZbDB7CusdRhgkD81VssRjQV5ZH13FbzCGcdnbbh6VwZ'
- '/ip4/127.0.0.1/tcp/8000/ws/p2p/16Uiu2HAm3TLea2NVs4dAqYM2gAgoV9CMKGeD1BkP3RAvmk7HBAbU'
- );
+ const peers = [
+ // '/ip4/127.0.0.1/tcp/8000/ws/p2p/16Uiu2HAm3TLea2NVs4dAqYM2gAgoV9CMKGeD1BkP3RAvmk7HBAbU',
+ '/dns4/waku-test.bloxy.one/tcp/8095/wss/p2p/16Uiu2HAmSZbDB7CusdRhgkD81VssRjQV5ZH13FbzCGcdnbbh6VwZ',
+ '/dns4/waku.fryorcraken.xyz/tcp/8000/wss/p2p/16Uiu2HAmMRvhDHrtiHft1FTUYnn6cVA8AWVrTyLUayJJ3MWpUZDB',
+ '/dns4/ivansete.xyz/tcp/8000/wss/p2p/16Uiu2HAmDAHuJ8w9zgxVnhtFe8otWNJdCewPAerJJPbXJcn8tu4r'
+ ];
+ for (const peer of peers) {
+ try {
+ await node.dial(peer);
+ } catch (error) {
+ console.error(`Error dialing peer ${peer}:`, error);
+ }
+ }
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(window as any).waku = node;
connectionState.update((state) => ({
@@ -227,7 +236,11 @@ export async function joinLobby(): Promise {
if (messageData.from === id) {
return;
}
- if (messageData.messageType === LobbyMessageType.Ping || messageData.to === id) {
+ if (
+ messageData.messageType === LobbyMessageType.Ongoing ||
+ messageData.messageType === LobbyMessageType.Ping ||
+ messageData.to === id
+ ) {
processUpdates([{ peerId: messageData.from, message: messageData, sent: false }]);
}
console.log('Received lobby message:', messageData);
@@ -344,6 +357,26 @@ export async function joinLobby(): Promise {
subscription.unsubscribe([lobbyDecoder.contentTopic]);
}
resume(Effect.succeed(params));
+ } else if (event.newState === PeerState.Ongoing) {
+ console.log(`Match ongoing with ${event.peerId}`);
+
+ // if we sent a match, then start the simulation
+ // stop responding to lobby messages, as we should now be in a match
+ if (fiber) {
+ fiber();
+ }
+ if (pingFiber) {
+ Effect.runFork(Fiber.interrupt(pingFiber));
+ }
+ if (subscription) {
+ subscription.unsubscribe([lobbyDecoder.contentTopic]);
+ }
+ if (event.message.match) {
+ const params: MatchParams = event.message.match;
+ resume(Effect.succeed(params));
+ } else {
+ resume(Effect.fail(new Error('No match found')));
+ }
}
};
diff --git a/examples/sds-demo/src/routes/state-graph/+page.svelte b/examples/sds-demo/src/routes/state-graph/+page.svelte
index 139d5a5..ea71d84 100644
--- a/examples/sds-demo/src/routes/state-graph/+page.svelte
+++ b/examples/sds-demo/src/routes/state-graph/+page.svelte
@@ -1,7 +1,7 @@
-
diff --git a/examples/sds-demo/src/lib/components/StateGraph.svelte b/examples/sds-demo/src/lib/components/StateGraph.svelte
deleted file mode 100644
index b31f6a1..0000000
--- a/examples/sds-demo/src/lib/components/StateGraph.svelte
+++ /dev/null
@@ -1,75 +0,0 @@
-
-
-
-
+
+ {#if currentIdFilter.id}
+
-
-
- 
-
-
-
-
-
- {entry.messageId}
-
- {#if entry.retrievalHint}
-
- {bytesToHex(entry.retrievalHint)}
-
- {/if}
-
+ ID: {currentIdFilter.id}
+
- {/each}
+ {/if}
+
+
+ {#each filteredHistory as event, index}
+
+
+
+
+ {#if event}
+
+ {/each}
+
+
+ {/if}
+
+
+
+
+
+
+
+
+
+
+
+ {event.messageId}
+
+
- {#each grid as row, y}
- {#each row as item, x}
-
-
-
diff --git a/examples/sds-demo/src/lib/components/StateGraphSummary.svelte b/examples/sds-demo/src/lib/components/StateGraphSummary.svelte
index 104eeee..10133fb 100644
--- a/examples/sds-demo/src/lib/components/StateGraphSummary.svelte
+++ b/examples/sds-demo/src/lib/components/StateGraphSummary.svelte
@@ -5,6 +5,7 @@
import { MessageChannelEvent } from '@waku/sds';
import { eventColors, eventNames, currentIdFilter, matchesIdFilter } from '$lib/utils/event.svelte';
import type { MessageChannelEventObject } from '$lib/sds/stream';
+ import { hash } from '$lib/utils/hash';
let { channelId = null }: { channelId: string | null } = $props();
let unsubscribe: (() => void) | null = $state(null);
@@ -14,8 +15,7 @@
return;
}
if (event.type === MessageChannelEvent.SyncSent || event.type === MessageChannelEvent.SyncReceived) {
- event
- return;
+ event.payload.messageId = hash(event.payload.messageId + event.payload.causalHistory[0].messageId)
}
console.log('updating virtual grid', event);
update_virtual_grid(event);
diff --git a/examples/sds-demo/src/lib/utils/frequency.ts b/examples/sds-demo/src/lib/utils/frequency.ts
new file mode 100644
index 0000000..2873a93
--- /dev/null
+++ b/examples/sds-demo/src/lib/utils/frequency.ts
@@ -0,0 +1,38 @@
+import { Schedule } from "effect";
+
+import { Ref } from "effect";
+
+import { Effect } from "effect";
+
+
+export function frequency() {
+ return Effect.runPromise(
+ Effect.gen(function* () {
+ const listenCondition = yield* Ref.make(false);
+ // const queue = yield* initializeQueue;
+ yield* Effect.all(
+ [
+ // setup filter
+ // subscribe(listenCondition),
+ // // send messages
+ Effect.repeat(Effect.sync(() => {}), Schedule.spaced('2000 millis')),
+ // // Effect.repeat(takeAndSend, Schedule.spaced('2000 millis')),
+ // // periodic sync
+ // Effect.repeat(sendSync, Schedule.spaced('10000 millis')),
+ // // periodically process queue
+ // Effect.repeat(processQueue, Schedule.spaced('200 millis')),
+ // // periodically sweep buffers
+ // Effect.repeat(sweep, Schedule.spaced('2000 millis')),
+ // // periodically switch off filter to miss messages
+ Effect.repeat(
+ Ref.update(listenCondition, (listening) => !listening),
+ Schedule.spaced('2000 millis')
+ )
+ ],
+ {
+ concurrency: 'unbounded'
+ }
+ );
+ })
+ );
+}
diff --git a/examples/sds-demo/src/lib/waku/lobby.svelte.ts b/examples/sds-demo/src/lib/waku/lobby.svelte.ts
index c5d2ad7..0a0d843 100644
--- a/examples/sds-demo/src/lib/waku/lobby.svelte.ts
+++ b/examples/sds-demo/src/lib/waku/lobby.svelte.ts
@@ -1,4 +1,5 @@
import { Effect, Option, pipe, Stream } from 'effect';
+import type { MatchParams } from './waku.svelte';
// Define the type for state transition events
export interface StateTransitionDetail {
@@ -22,7 +23,8 @@ export enum LobbyMessageType {
Ping = 'ping',
Request = 'request',
Accept = 'accept',
- Match = 'match'
+ Match = 'match',
+ Ongoing = 'ongoing'
}
enum LobbyEvent {
@@ -32,7 +34,8 @@ enum LobbyEvent {
GotAccept = 'got_accept',
SentAccept = 'sent_accept',
GotMatch = 'got_match',
- SentMatch = 'sent_match'
+ SentMatch = 'sent_match',
+ Ongoing = 'ongoing'
}
export type LobbyMessage = {
@@ -41,6 +44,7 @@ export type LobbyMessage = {
expiry?: Date;
from: string;
to?: string;
+ match?: MatchParams;
};
export enum PeerState {
@@ -51,25 +55,26 @@ export enum PeerState {
AcceptTo = 'accept_to',
AcceptFrom = 'accept_from',
Success = 'success',
- Failure = 'failure'
+ Failure = 'failure',
+ Ongoing = 'ongoing'
}
// Create a class that extends EventTarget for native event handling
class LobbyState extends EventTarget {
// The peer state map
peerState = $state(new Map
- {#if item !== null && y * columns + x < identicons.length}
-
- {/each}
- {/each}
-