openvibers/openvibe.examples Commits
Examples: the proofs read Events by cursor only (no seq) (#19)
Changes
@@@ -212,7 +212,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODUconst cursorPath = path.join(dir, 'cursor.json');const seen = [];const s = createSubscriber(subscriberConfig({ ...creds(), OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {-- fetch: fetchImpl, log: { log: () => {}, error: (m) => say(m) }, onEvent: (event, { seq }) => { seen.push({ seq, id: event.event_id }); },++ fetch: fetchImpl, log: { log: () => {}, error: (m) => say(m) }, onEvent: (event) => { seen.push({ id: event.event_id }); },});// Start the subscriber at the head (an opaque cursor), then publish: the pull finds what came after it.await s.startAtHead();@@@ -224,7 +224,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODUconst n = await s.pullOnce();const mine = seen.find((e) => e.id === pub.event_id);if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull of ${topics[0]}`);-- say(`pulled ${topics[0]} with events.app.read: ${n} event(s), found ${mine.id} at seq ${mine.seq}`);++ say(`pulled ${topics[0]} with events.app.read: ${n} event(s), found ${mine.id}`);});await step('export', async () => {@@@ -69,7 +69,7 @@ async function main() {const { createUploader, explain } = require(path.join(ROOT, 'examples/media-uploader/upload'));const { projectKey } = require('openvibe-sdk/events');-- const { createSubscriber, createCursorStore, loadConfig: subscriberConfig } = require(path.join(ROOT, 'examples/event-subscriber/subscriber'));++ const { createSubscriber, loadConfig: subscriberConfig } = require(path.join(ROOT, 'examples/event-subscriber/subscriber'));const { publishAppEvent, loadConfig: publishConfig } = require(path.join(ROOT, 'examples/event-subscriber/publish'));const { createJobRunner, loadConfig: jobConfig } = require(path.join(ROOT, 'examples/tool-job/run-job'));const { loadConfig: mediaConfig } = require(path.join(ROOT, 'examples/media-uploader/upload'));@@@ -108,21 +108,22 @@ async function main() {});await step('event-subscriber: publish to the own topic, then pull it', async () => {-- const pub = await publishAppEvent(publishConfig(base), { name: 'e2e.ran', subject: { type: 'check', id: 'e2e' }, payload: { at: new Date().toISOString() } });-- console.log(` published ${pub.event_type} ${pub.event_id} -> seq ${pub.seq}${pub.duplicate ? ' (duplicate)' : ''}`);const cursorPath = path.join(dir, 'cursor.json');-- createCursorStore(cursorPath).set('pull', Math.max(0, pub.seq - 1));const seen = [];const s = createSubscriber(subscriberConfig({ ...base, OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {-- log, onEvent: (event, { seq }) => { seen.push({ seq, id: event.event_id, type: event.event_type }); },++ log, onEvent: (event) => { seen.push({ id: event.event_id, type: event.event_type }); },});++ // Start at the head (an opaque cursor), then publish: the pull finds what came after it.++ await s.startAtHead();++ const pub = await publishAppEvent(publishConfig(base), { name: 'e2e.ran', subject: { type: 'check', id: 'e2e' }, payload: { at: new Date().toISOString() } });++ console.log(` published ${pub.event_type} ${pub.event_id}${pub.duplicate ? ' (duplicate)' : ''}`);const topics = await s.resolveTopics();if (topics[0] !== `app.${projectKey(env.OV_PROJECT_ID)}.*`) throw new Error(`unexpected topic ${topics[0]}`);const n = await s.pullOnce();-- console.log(` pulled ${topics.join(', ')} from seq ${pub.seq - 1}: ${n} event(s), cursor now ${createCursorStore(cursorPath).get('pull')}`);++ console.log(` pulled ${topics.join(', ')} from the head: ${n} event(s)`);const mine = seen.find((e) => e.id === pub.event_id);if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull`);-- console.log(` found ${mine.id} at seq ${mine.seq}`);++ console.log(` found ${mine.id}`);});await step('tool-job: img.process convert to WebP', async () => {@@@ -166,8 +166,8 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, clientlet moved;do {page = await st.events.pull({ topic: 'tools.job.*', after, limit: 200 });-- for (const { seq, event } of page.events || []) {-- if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, seq);++ for (const { event } of page.events || []) {++ if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, event.event_id);}moved = Boolean(page.next_cursor) && page.next_cursor !== after;if (moved) after = page.next_cursor;@@@ -176,7 +176,7 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, clientif (Date.now() >= deadline) throw new Error(`not in the Events store after ${eventsWaitMs / 1000} s: ${EVENT_TYPES.filter((t) => !found.has(t)).join(', ')}`);await sleep(pollMs);}-- say(EVENT_TYPES.map((t) => `${t} seq ${found.get(t)}`).join(', '));++ say(EVENT_TYPES.map((t) => `${t} ${found.get(t)}`).join(', '));});const ok = steps.every((s) => s.ok);
@@ -212,7 +212,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODU
@@ -212,7 +212,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODU
const cursorPath = path.join(dir, 'cursor.json');
const cursorPath = path.join(dir, 'cursor.json');
const seen = [];
const seen = [];
const s = createSubscriber(subscriberConfig({ ...creds(), OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {
const s = createSubscriber(subscriberConfig({ ...creds(), OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {
- fetch: fetchImpl, log: { log: () => {}, error: (m) => say(m) }, onEvent: (event, { seq }) => { seen.push({ seq, id: event.event_id }); },
+ fetch: fetchImpl, log: { log: () => {}, error: (m) => say(m) }, onEvent: (event) => { seen.push({ id: event.event_id }); },
});
});
// Start the subscriber at the head (an opaque cursor), then publish: the pull finds what came after it.
// Start the subscriber at the head (an opaque cursor), then publish: the pull finds what came after it.
await s.startAtHead();
await s.startAtHead();
@@ -224,7 +224,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODU
@@ -224,7 +224,7 @@ async function runDeveloperPath({ network = PRODUCTION_NETWORK, services = PRODU
const n = await s.pullOnce();
const n = await s.pullOnce();
const mine = seen.find((e) => e.id === pub.event_id);
const mine = seen.find((e) => e.id === pub.event_id);
if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull of ${topics[0]}`);
if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull of ${topics[0]}`);
- say(`pulled ${topics[0]} with events.app.read: ${n} event(s), found ${mine.id} at seq ${mine.seq}`);
+ say(`pulled ${topics[0]} with events.app.read: ${n} event(s), found ${mine.id}`);
});
});
await step('export', async () => {
await step('export', async () => {
@@ -69,7 +69,7 @@ async function main() {
@@ -69,7 +69,7 @@ async function main() {
const { createUploader, explain } = require(path.join(ROOT, 'examples/media-uploader/upload'));
const { createUploader, explain } = require(path.join(ROOT, 'examples/media-uploader/upload'));
const { projectKey } = require('openvibe-sdk/events');
const { projectKey } = require('openvibe-sdk/events');
- const { createSubscriber, createCursorStore, loadConfig: subscriberConfig } = require(path.join(ROOT, 'examples/event-subscriber/subscriber'));
+ const { createSubscriber, loadConfig: subscriberConfig } = require(path.join(ROOT, 'examples/event-subscriber/subscriber'));
const { publishAppEvent, loadConfig: publishConfig } = require(path.join(ROOT, 'examples/event-subscriber/publish'));
const { publishAppEvent, loadConfig: publishConfig } = require(path.join(ROOT, 'examples/event-subscriber/publish'));
const { createJobRunner, loadConfig: jobConfig } = require(path.join(ROOT, 'examples/tool-job/run-job'));
const { createJobRunner, loadConfig: jobConfig } = require(path.join(ROOT, 'examples/tool-job/run-job'));
const { loadConfig: mediaConfig } = require(path.join(ROOT, 'examples/media-uploader/upload'));
const { loadConfig: mediaConfig } = require(path.join(ROOT, 'examples/media-uploader/upload'));
@@ -108,21 +108,22 @@ async function main() {
@@ -108,21 +108,22 @@ async function main() {
});
});
await step('event-subscriber: publish to the own topic, then pull it', async () => {
await step('event-subscriber: publish to the own topic, then pull it', async () => {
- const pub = await publishAppEvent(publishConfig(base), { name: 'e2e.ran', subject: { type: 'check', id: 'e2e' }, payload: { at: new Date().toISOString() } });
- console.log(` published ${pub.event_type} ${pub.event_id} -> seq ${pub.seq}${pub.duplicate ? ' (duplicate)' : ''}`);
const cursorPath = path.join(dir, 'cursor.json');
const cursorPath = path.join(dir, 'cursor.json');
- createCursorStore(cursorPath).set('pull', Math.max(0, pub.seq - 1));
const seen = [];
const seen = [];
const s = createSubscriber(subscriberConfig({ ...base, OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {
const s = createSubscriber(subscriberConfig({ ...base, OV_EVENTS_MODE: 'pull', OV_CURSOR_PATH: cursorPath }), {
- log, onEvent: (event, { seq }) => { seen.push({ seq, id: event.event_id, type: event.event_type }); },
+ log, onEvent: (event) => { seen.push({ id: event.event_id, type: event.event_type }); },
});
});
+ // Start at the head (an opaque cursor), then publish: the pull finds what came after it.
+ await s.startAtHead();
+ const pub = await publishAppEvent(publishConfig(base), { name: 'e2e.ran', subject: { type: 'check', id: 'e2e' }, payload: { at: new Date().toISOString() } });
+ console.log(` published ${pub.event_type} ${pub.event_id}${pub.duplicate ? ' (duplicate)' : ''}`);
const topics = await s.resolveTopics();
const topics = await s.resolveTopics();
if (topics[0] !== `app.${projectKey(env.OV_PROJECT_ID)}.*`) throw new Error(`unexpected topic ${topics[0]}`);
if (topics[0] !== `app.${projectKey(env.OV_PROJECT_ID)}.*`) throw new Error(`unexpected topic ${topics[0]}`);
const n = await s.pullOnce();
const n = await s.pullOnce();
- console.log(` pulled ${topics.join(', ')} from seq ${pub.seq - 1}: ${n} event(s), cursor now ${createCursorStore(cursorPath).get('pull')}`);
+ console.log(` pulled ${topics.join(', ')} from the head: ${n} event(s)`);
const mine = seen.find((e) => e.id === pub.event_id);
const mine = seen.find((e) => e.id === pub.event_id);
if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull`);
if (!mine) throw new Error(`the published event ${pub.event_id} was not in the pull`);
- console.log(` found ${mine.id} at seq ${mine.seq}`);
+ console.log(` found ${mine.id}`);
});
});
await step('tool-job: img.process convert to WebP', async () => {
await step('tool-job: img.process convert to WebP', async () => {
@@ -166,8 +166,8 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, client
@@ -166,8 +166,8 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, client
let moved;
let moved;
do {
do {
page = await st.events.pull({ topic: 'tools.job.*', after, limit: 200 });
page = await st.events.pull({ topic: 'tools.job.*', after, limit: 200 });
- for (const { seq, event } of page.events || []) {
- if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, seq);
+ for (const { event } of page.events || []) {
+ if (event && event.subject && event.subject.id === st.jobId && EVENT_TYPES.includes(event.event_type)) found.set(event.event_type, event.event_id);
}
}
moved = Boolean(page.next_cursor) && page.next_cursor !== after;
moved = Boolean(page.next_cursor) && page.next_cursor !== after;
if (moved) after = page.next_cursor;
if (moved) after = page.next_cursor;
@@ -176,7 +176,7 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, client
@@ -176,7 +176,7 @@ async function runToolsJobProof({ network = PRODUCTION_NETWORK, toolsUrl, client
if (Date.now() >= deadline) throw new Error(`not in the Events store after ${eventsWaitMs / 1000} s: ${EVENT_TYPES.filter((t) => !found.has(t)).join(', ')}`);
if (Date.now() >= deadline) throw new Error(`not in the Events store after ${eventsWaitMs / 1000} s: ${EVENT_TYPES.filter((t) => !found.has(t)).join(', ')}`);
await sleep(pollMs);
await sleep(pollMs);
}
}
- say(EVENT_TYPES.map((t) => `${t} seq ${found.get(t)}`).join(', '));
+ say(EVENT_TYPES.map((t) => `${t} ${found.get(t)}`).join(', '));
});
});
const ok = steps.every((s) => s.ok);
const ok = steps.every((s) => s.ok);