Skip to content

Commit 8b2a761

Browse files
joaoantoniocardosojoaoantoniocardoso
committed
core: frontend: Adapt to the new Zenoh API
1 parent 4dd3cee commit 8b2a761

6 files changed

Lines changed: 64 additions & 43 deletions

File tree

‎core/frontend/src/components/cloud/CloudTrayMenu.vue‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -174,7 +174,7 @@
174174

175175
<script lang="ts">
176176
import {
177-
QueryTarget, Receiver, RecvErr, ReplyError, Sample, Session, Subscriber,
177+
QueryTarget, ReplyError, Sample, Session, Subscriber,
178178
} from '@eclipse-zenoh/zenoh-ts'
179179
import axios from 'axios'
180180
import { StatusCodes } from 'http-status-codes'
@@ -191,7 +191,7 @@ import {
191191
import SpinningLogo from '@/components/common/SpinningLogo.vue'
192192
import PullProgress from '@/components/utils/PullProgress.vue'
193193
import filebrowser from '@/libs/filebrowser'
194-
import zenoh from '@/libs/zenoh'
194+
import zenoh, { receiveQueryReply, RecvErr } from '@/libs/zenoh'
195195
import bag from '@/store/bag'
196196
import { FilebrowserFile } from '@/types/filebrowser'
197197
import { InstalledExtensionData, RunningContainer } from '@/types/kraken'
@@ -447,30 +447,30 @@ export default Vue.extend({
447447
}
448448
449449
try {
450-
const receiver: void | Receiver = this.file_sync_session.get(topic, {
450+
const receiver = await this.file_sync_session.get(topic, {
451451
target: QueryTarget.BestMatching,
452452
})
453453
454-
if (!(receiver instanceof Receiver)) {
454+
if (!receiver) {
455455
console.warn(`[CloudTrayMenu] Query for ${topic} returned void receiver.`)
456456
return null
457457
}
458458
459-
let reply = await receiver.receive()
459+
let reply = await receiveQueryReply(receiver)
460460
while (reply !== RecvErr.Disconnected) {
461461
if (reply === RecvErr.MalformedReply) {
462462
console.warn(`[CloudTrayMenu] Malformed reply while querying ${topic}.`)
463463
} else {
464464
const response = reply.result()
465465
if (response instanceof Sample) {
466-
const payload = response.payload().to_string()
466+
const payload = response.payload().toString()
467467
return this.parseFileSyncQueue(payload)
468468
}
469469
const errorResponse: ReplyError = response
470-
console.warn(`[CloudTrayMenu] Query error for ${topic}:`, errorResponse.payload().to_string())
470+
console.warn(`[CloudTrayMenu] Query error for ${topic}:`, errorResponse.payload().toString())
471471
}
472472
473-
reply = await receiver.receive()
473+
reply = await receiveQueryReply(receiver)
474474
}
475475
} catch (error) {
476476
console.error(`[CloudTrayMenu] Failed to query ${topic}:`, error)
@@ -518,13 +518,12 @@ export default Vue.extend({
518518
}
519519
520520
try {
521-
this.uploading_subscriber = await this.file_sync_session.declare_subscriber(
521+
this.uploading_subscriber = await this.file_sync_session.declareSubscriber(
522522
MAJOR_TOM_FILE_SYNC_UPLOADING_TOPIC,
523523
{
524524
handler: (sample: Sample) => {
525525
this.handleUploadingSample(sample)
526526
},
527-
history: true,
528527
},
529528
)
530529
} catch (error) {
@@ -533,7 +532,7 @@ export default Vue.extend({
533532
},
534533
handleUploadingSample(sample: Sample): void {
535534
try {
536-
const payload = sample.payload().to_string()
535+
const payload = sample.payload().toString()
537536
const event = JSON.parse(payload) as FileSyncUploadingEvent
538537
const transfer = this.normalizeUploadingEvent(event)
539538
if (!transfer) {

‎core/frontend/src/components/kraken/modals/ExtensionLogsModal.vue‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,7 @@ export default Vue.extend({
214214
this.modal_subscriber = await kraken.createExtensionLogsSubscriber(topic, this.handleSubscriber)
215215
},
216216
async handleSubscriber(sample: Sample) {
217-
const payloadString = sample.payload().to_string()
217+
const payloadString = sample.payload().toString()
218218
219219
let message = payloadString
220220
try {

‎core/frontend/src/components/zenoh-inspector/ZenohInspector.vue‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,7 @@ export default Vue.extend({
182182
if (!this.current_message?.payload || !this.video_reader) {
183183
return null
184184
}
185-
const msg: { data: Uint8Array } = this.video_reader.readMessage(this.current_message.payload.to_bytes())
185+
const msg: { data: Uint8Array } = this.video_reader.readMessage(this.current_message.payload.toBytes())
186186
return msg.data
187187
},
188188
},
@@ -215,12 +215,12 @@ export default Vue.extend({
215215
}
216216
217217
if (message.encoding === Encoding.TEXT_PLAIN.toString()) {
218-
formattedMessage.payload = message.payload.to_string()
218+
formattedMessage.payload = message.payload.toString()
219219
} else if (message.encoding === Encoding.APPLICATION_JSON.toString()) {
220-
formattedMessage.payload = JSON.parse(message.payload.to_string())
220+
formattedMessage.payload = JSON.parse(message.payload.toString())
221221
} else if (message.encoding === Encoding.ZENOH_BYTES.toString()) {
222222
try {
223-
formattedMessage.payload = JSON.parse(message.payload.to_string())
223+
formattedMessage.payload = JSON.parse(message.payload.toString())
224224
} catch (exception) {
225225
// Keep the raw payload if it's not valid JSON
226226
formattedMessage.payload = message.payload.toString()
@@ -234,7 +234,7 @@ export default Vue.extend({
234234
this.session = await zenoh.getSession()
235235
236236
// Setup regular message subscriber
237-
this.subscriber = await this.session.declare_subscriber('**', {
237+
this.subscriber = await this.session.declareSubscriber('**', {
238238
handler: async (sample: Sample) => {
239239
const topic = sample.keyexpr().toString()
240240
const payload = sample.payload()
@@ -261,7 +261,7 @@ export default Vue.extend({
261261
// Setup liveliness subscriber
262262
const lv_ke = '@/**/@ros2_lv/**'
263263
264-
this.liveliness_subscriber = await this.session.liveliness().declare_subscriber(lv_ke, {
264+
this.liveliness_subscriber = await this.session.liveliness().declareSubscriber(lv_ke, {
265265
handler: (sample: Sample) => {
266266
// Parse the liveliness token using regex
267267
// eslint-disable-next-line max-len

‎core/frontend/src/components/zenoh-inspector/ZenohNetwork.vue‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -57,13 +57,13 @@
5757

5858
<script lang="ts">
5959
import {
60-
QueryTarget, Receiver, RecvErr, ReplyError, Sample, Session,
60+
QueryTarget, ReplyError, Sample, Session,
6161
} from '@eclipse-zenoh/zenoh-ts'
6262
import cytoscape, { Core } from 'cytoscape'
6363
import fcose, { FcoseLayoutOptions } from 'cytoscape-fcose'
6464
import Vue from 'vue'
6565
66-
import zenoh from '@/libs/zenoh'
66+
import zenoh, { receiveQueryReply, RecvErr } from '@/libs/zenoh'
6767
6868
interface NetworkNode {
6969
id: string
@@ -295,16 +295,16 @@ export default Vue.extend({
295295
},
296296
async queryRouters() {
297297
try {
298-
const receiver: void | Receiver = this.session!.get('@/*/router', {
298+
const receiver = await this.session!.get('@/*/router', {
299299
target: QueryTarget.BestMatching,
300300
})
301301
302-
if (!(receiver instanceof Receiver)) {
302+
if (!receiver) {
303303
console.warn('[Zenoh Network] Router query returned void')
304304
return
305305
}
306306
307-
let reply = await receiver.receive()
307+
let reply = await receiveQueryReply(receiver)
308308
let responseCount = 0
309309
310310
while (reply !== RecvErr.Disconnected && responseCount < 10) { // Limit responses to prevent infinite loops
@@ -317,7 +317,7 @@ export default Vue.extend({
317317
responseCount += 1
318318
319319
try {
320-
const payload = sample.payload().to_string()
320+
const payload = sample.payload().toString()
321321
console.debug('[Zenoh Network] Router payload:', payload)
322322
323323
const data = JSON.parse(payload)
@@ -369,27 +369,27 @@ export default Vue.extend({
369369
}
370370
} else {
371371
const replyError: ReplyError = resp
372-
console.error('[Zenoh Network] Router query error:', replyError.payload().to_string())
372+
console.error('[Zenoh Network] Router query error:', replyError.payload().toString())
373373
}
374374
}
375-
reply = await receiver.receive()
375+
reply = await receiveQueryReply(receiver)
376376
}
377377
} catch (error) {
378378
console.warn('[Zenoh Network] Router query failed:', error)
379379
}
380380
},
381381
async queryPeers() {
382382
try {
383-
const receiver: void | Receiver = this.session!.get('@/*/peer', {
383+
const receiver = await this.session!.get('@/*/peer', {
384384
target: QueryTarget.BestMatching,
385385
})
386386
387-
if (!(receiver instanceof Receiver)) {
387+
if (!receiver) {
388388
console.warn('[Zenoh Network] Peer query returned void')
389389
return
390390
}
391391
392-
let reply = await receiver.receive()
392+
let reply = await receiveQueryReply(receiver)
393393
let responseCount = 0
394394
395395
while (reply !== RecvErr.Disconnected && responseCount < 10) { // Limit responses
@@ -402,7 +402,7 @@ export default Vue.extend({
402402
responseCount += 1
403403
404404
try {
405-
const payload = sample.payload().to_string()
405+
const payload = sample.payload().toString()
406406
const data = JSON.parse(payload)
407407
const peerId = data.zid || data.id || 'unknown-peer'
408408
if (!this.networkData.nodes.find((n) => n.id === peerId)) {
@@ -416,10 +416,10 @@ export default Vue.extend({
416416
}
417417
} else {
418418
const replyError: ReplyError = resp
419-
console.error('[Zenoh Network] Peer query error:', replyError.payload().to_string())
419+
console.error('[Zenoh Network] Peer query error:', replyError.payload().toString())
420420
}
421421
}
422-
reply = await receiver.receive()
422+
reply = await receiveQueryReply(receiver)
423423
}
424424
} catch (error) {
425425
console.warn('[Zenoh Network] Peer query failed:', error)
@@ -430,16 +430,16 @@ export default Vue.extend({
430430
try {
431431
const keyexpr = `@/${zid}/${whatami}`
432432
433-
const receiver: void | Receiver = this.session!.get(keyexpr, {
433+
const receiver = await this.session!.get(keyexpr, {
434434
target: QueryTarget.BestMatching,
435435
})
436436
437-
if (!(receiver instanceof Receiver)) {
437+
if (!receiver) {
438438
console.warn(`[Zenoh Network] Specific node query for ${keyexpr} returned void`)
439439
return null
440440
}
441441
442-
const reply = await receiver.receive()
442+
const reply = await receiveQueryReply(receiver)
443443
if (reply === RecvErr.Disconnected) {
444444
console.debug(`[Zenoh Network] No response for ${keyexpr}`)
445445
return null
@@ -454,7 +454,7 @@ export default Vue.extend({
454454
if (resp instanceof Sample) {
455455
const sample: Sample = resp
456456
try {
457-
const payload = sample.payload().to_string()
457+
const payload = sample.payload().toString()
458458
console.debug(`[Zenoh Network] Specific node response for ${keyexpr}:`, payload)
459459
const data = JSON.parse(payload)
460460
@@ -469,7 +469,7 @@ export default Vue.extend({
469469
}
470470
} else {
471471
const replyError: ReplyError = resp
472-
console.error(`[Zenoh Network] Specific node query error for ${keyexpr}:`, replyError.payload().to_string())
472+
console.error(`[Zenoh Network] Specific node query error for ${keyexpr}:`, replyError.payload().toString())
473473
return null
474474
}
475475
} catch (error) {

‎core/frontend/src/libs/console-logger.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -145,7 +145,12 @@ class ConsoleLogger {
145145
const topic = `frontend/${frontend.frontend_id}/logs`
146146
const payload = JSON.stringify(message)
147147

148+
// put() is async in zenoh 1.9; swallow rejections via originalConsole so a failed publish
149+
// cannot surface as an unhandledrejection and re-enter publishMessage (infinite feedback loop).
148150
this.session.put(topic, payload, { encoding: ConsoleLogger.LOG_ENCODING })
151+
.catch((publishError) => {
152+
this.originalConsole.error('[ConsoleLogger] Failed to publish message:', publishError)
153+
})
149154
} catch (error) {
150155
this.originalConsole.error('[ConsoleLogger] Failed to publish message:', error)
151156
}

‎core/frontend/src/libs/zenoh/index.ts‎

Lines changed: 23 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,21 @@
1-
import { Config, QueryTarget, Receiver, RecvErr, Sample, Session, Subscriber } from '@eclipse-zenoh/zenoh-ts'
1+
import { ChannelReceiver, Config, QueryTarget, Reply, Sample, Session, Subscriber } from '@eclipse-zenoh/zenoh-ts'
2+
3+
// zenoh-ts 1.9 no longer exports Receiver/RecvErr; the get() channel rejects receive() on close
4+
// (and the query has its own timeout), so we just map that rejection to Disconnected.
5+
export enum RecvErr {
6+
Disconnected,
7+
MalformedReply,
8+
}
9+
10+
export type Receiver = ChannelReceiver<Reply>
11+
12+
export async function receiveQueryReply(receiver: Receiver): Promise<Reply | RecvErr> {
13+
try {
14+
return await receiver.receive()
15+
} catch {
16+
return RecvErr.Disconnected
17+
}
18+
}
219

320
class ZenohManager {
421
private static instance: ZenohManager
@@ -38,11 +55,11 @@ class ZenohManager {
3855
return null
3956
}
4057

41-
const receiver: Receiver | void = session.get(key, {
58+
const receiver = await session.get(key, {
4259
target,
4360
})
4461

45-
if (!(receiver instanceof Receiver)) {
62+
if (!receiver) {
4663
console.error('Failed to create query receiver. No queryable found or connection error.')
4764
return null
4865
}
@@ -51,7 +68,7 @@ class ZenohManager {
5168
setTimeout(() => resolve(null), timeout)
5269
})
5370

54-
const replyPromise = receiver.receive()
71+
const replyPromise = receiveQueryReply(receiver)
5572
const reply = await Promise.race([replyPromise, timeoutPromise])
5673

5774
if (reply === null || reply === RecvErr.Disconnected) {
@@ -67,7 +84,7 @@ class ZenohManager {
6784

6885
const payload = (reply as { result: () => Sample }).result()
6986
try {
70-
return JSON.parse(payload.payload().to_string())
87+
return JSON.parse(payload.payload().toString())
7188
} catch (error) {
7289
console.error('Error parsing response:', error)
7390
return null
@@ -81,7 +98,7 @@ class ZenohManager {
8198
return null
8299
}
83100

84-
return session.declare_subscriber(topic, {
101+
return session.declareSubscriber(topic, {
85102
handler,
86103
})
87104
}

0 commit comments

Comments
 (0)