Skip to content

Re-subscribe on event/observed property error #34

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pierre-josselin wants to merge 1 commit into
base: main
Choose a base branch
from
+101 −110
Open

Re-subscribe on event/observed property error #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 49 additions & 58 deletions node-red-node-wot/src/wot-event.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ module.exports = function (RED) {
let node = this
let consumedThing
let subscription
let repeatId

this.status({})

Expand All @@ -20,81 +19,73 @@ module.exports = function (RED) {

const thingNode = RED.nodes.getNode(config.thing)
thingNode.addUpdateTDListener(async (_consumedThing) => {
if (repeatId) {
clearInterval(repeatId)
repeatId = undefined
}
if (subscription) {
// Stop if already subscribed
await subscription.stop()
}
subscription = undefined
consumedThing = _consumedThing
// Repeat until event subscription succeeds.
await new Promise((resolve) => {
repeatId = setInterval(() => {
consumedThing
.subscribeEvent(
config.event,
async (resp) => {
if (resp) {
let payload
try {
payload = await resp.value()
} catch (err) {
node.error(`[error] failed to get event. err: ${err.toString()}`)
console.error(`[error] failed to get event. err: `, err)
while (true) {
try {
await new Promise((resolve, reject) => {
consumedThing
.subscribeEvent(
config.event,
async (resp) => {
if (resp) {
let payload
try {
payload = await resp.value()
} catch (err) {
node.error(`[error] failed to get event. err: ${err.toString()}`)
console.error(`[error] failed to get event. err: `, err)
}
node.send({ payload, topic: config.topic })
}
node.send({ payload, topic: config.topic })
node.status({
fill: "green",
shape: "dot",
text: "Subscribed",
})
},
(err) => {
console.error("[error] subscribe events.", err)
node.error(`[error] subscribe events. err: ${err.toString()}`)
node.status({
fill: "red",
shape: "ring",
text: "Subscription error",
})
reject(err)
}
)
.then((sub) => {
subscription = sub
node.status({
fill: "green",
shape: "dot",
text: "Subscribed",
})
},
(err) => {
console.error("[error] subscribe events.", err)
node.error(`[error] subscribe events. err: ${err.toString()}`)
node.status({
fill: "red",
shape: "ring",
text: "Subscription error",
})
},
() => {
console.error("[warn] Subscription ended.")
node.warn("[warn] Subscription ended.")
node.status({})
subscription = undefined
}
)
.then((sub) => {
subscription = sub
clearInterval(repeatId)
repeatId = undefined
resolve()
})
.catch((err) => {
console.warn("[warn] event subscribe error. try again. error: " + err)
})
}, 1000)
})

if (subscription) {
node.status({
fill: "green",
shape: "dot",
text: "Subscribed",
})
})
.catch(reject)
});
} catch (err) {
console.warn("[warn] event subscribe error. try again. error: " + err)
node.status({
fill: "red",
shape: "ring",
text: "Subscription error",
})
}
if (subscription) {
await subscription.stop()
}
await new Promise((resolve) => setTimeout(resolve, 1000))
}
})

this.on("close", async function (removed, done) {
if (repeatId) {
clearInterval(repeatId)
repeatId = undefined
}
if (subscription) {
// Stop if already subscribed
await subscription.stop()
Expand Down
104 changes: 52 additions & 52 deletions node-red-node-wot/src/wot-property.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ module.exports = function (RED) {
let node = this
let consumedThing
let subscription
let repeatId

this.status({})

Expand All @@ -24,10 +23,6 @@ module.exports = function (RED) {

const thingNode = RED.nodes.getNode(config.thing)
thingNode.addUpdateTDListener(async (_consumedThing) => {
if (repeatId) {
clearInterval(repeatId)
repeatId = undefined
}
if (subscription) {
// Stop if already subscribed
await subscription.stop()
Expand All @@ -38,53 +33,62 @@ module.exports = function (RED) {
return
}
// Repeat until observeProperty succeeds.
await new Promise((resolve) => {
repeatId = setInterval(() => {
consumedThing
.observeProperty(
config.property,
async (resp) => {
let payload
try {
payload = await resp.value()
} catch (err) {
node.error(`[error] failed to get property change. err: ${err.toString()}`)
console.error(`[error] failed to get property change. err:`, err)
while (true) {
try {
await new Promise((resolve, reject) => {
consumedThing
.observeProperty(
config.property,
async (resp) => {
if (resp) {
let payload
try {
payload = await resp.value()
} catch (err) {
node.error(`[error] failed to get property change. err: ${err.toString()}`)
console.error(`[error] failed to get property change. err:`, err)
}
node.send({ payload, topic: config.topic })
}
node.status({
fill: "green",
shape: "dot",
text: "Subscribed",
})
},
(err) => {
node.error(`[error] property observe error. error: ${err.toString()}`)
console.error(`[error] property observe error. error: `, err)
node.status({
fill: "red",
shape: "ring",
text: "Observe error",
})
reject(err)
}
node.send({ payload, topic: config.topic })
},
(err) => {
node.error(`[error] property observe error. error: ${err.toString()}`)
console.error(`[error] property observe error. error: `, err)
)
.then((sub) => {
subscription = sub
node.status({
fill: "red",
shape: "ring",
text: "Observe error",
fill: "green",
shape: "dot",
text: "connected",
})
}
)
.then((sub) => {
subscription = sub
clearInterval(repeatId)
repeatId = undefined
resolve()
})
.catch((err) => {
console.warn("[warn] property observe error. try again. error: " + err)
node.status({
fill: "red",
shape: "ring",
text: "Observe error",
})
})
}, 1000)
})
if (subscription) {
node.status({
fill: "green",
shape: "dot",
text: "connected",
})
.catch(reject)
});
} catch (err) {
console.warn("[warn] property observe error. try again. error: " + err)
node.status({
fill: "red",
shape: "ring",
text: "Observe error",
})
}
if (subscription) {
await subscription.stop()
}
await new Promise((resolve) => setTimeout(resolve, 1000))
}
})

Expand Down Expand Up @@ -126,10 +130,6 @@ module.exports = function (RED) {
})

node.on("close", async function (removed, done) {
if (repeatId) {
clearInterval(repeatId)
repeatId = undefined
}
if (subscription) {
// Stop if already subscribed
await subscription.stop()
Expand Down
Loading