Changefeeds 是否具有相同的过滤器也具有相同的光标?
Do changefeeds has same filter is also has same cursor?
代码:
const WebSocket = require('ws')
const r = require('rethinkdb')
async function start () {
try {
/**
* Connect database
*/
const conn = await r.connect()
conn.use('app')
/**
* Create socket server
*/
const wss = new WebSocket.Server({ port: 8082 })
wss.on('connection', (client) => {
/**
* Client setting subscribe rethinkdb data
* WARNING : doesn't validate payload
*/
client.on('message', (payload) => {
const filterData = JSON.parse(payload)
/**
* Send all data base on filter
*
/**
* Subscribe rethink data base on message request
* Send all data base on update
*/
r.table('foo')
.changes({
'includeTypes': true,
'includeInitial': true,
"squash": true // Squash many changes into one event
})
.run(conn, async (err, cursor) => {
if (err) {
throw err
}
client.cursor = cursor
cursor.each((err, row) => {
if (err) {
throw err
}
/**
* Check if client is connected
* Else close cursor
*
* Also useful for check if connect is dead
*/
if (client.readyState !== client.OPEN) {
client.terminate()
client.cursor.close()
return
}
client.send(JSON.stringify(row))
})
/**
* Close stream when :
* Client disconnect
* Client send new message
*/
client.on('message', () => {
cursor.close()
})
})
})
})
} catch (err) {
console.log(err)
}
}
start()
错误
ReqlDriverError: Cursor is closed.
at ReqlDriverError.ReqlError [as constructor] (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\errors.js:23:13)
at new ReqlDriverError (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\errors.js:68:50)
at Feed.IterableResult._promptNext (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:128:10)
at d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:200:22
at tryCatcher (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\util.js:26:23)
at Function.Promise.fromNode (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:168:30)
at Feed.<anonymous> (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:203:20)
at Feed.<anonymous> (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\util.js:43:16)
at d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:262:24
at tryCatcher (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\util.js:26:23)
at Promise.successAdapter (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\nodeify.js:23:30)
at Promise._settlePromiseAt (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:582:21)
at Promise._settlePromises (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:700:14)
at Async._drainQueue (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:123:16)
at Async._drainQueues (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:133:10)
at Immediate.Async.drainQueues (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:15:14)
重现问题::
- 连接一个
- 重新连接(之前的连接已经关闭,但它的事件还没有关闭)
- 在rethinkdb中更改数据(此时我在if语句中打了个断点)
- 第一个连接关闭 -> 开始关闭游标并停止执行下一个
- 第二个还没关闭
- 错误:游标已关闭
问题:
changefeeds 是否具有相同的过滤器也具有相同的光标?每个 websocket 客户端都应该有自己的 changefeed 游标。为什么我关闭了第一个的套接字,然后第二个也关闭了?
改变这个
if (client.readyState !== client.OPEN) {
client.terminate()
client.cursor.close()
return
}
client.send(JSON.stringify(row))
进入这个:
if (client.readyState === client.OPEN) {
client.send(JSON.stringify(row))
} else {
await cursor.close()
client.terminate()
}
解决我的问题。我仍然收到错误消息:ReqlDriverError: Cursor is closed. in try catch...
代码:
const WebSocket = require('ws')
const r = require('rethinkdb')
async function start () {
try {
/**
* Connect database
*/
const conn = await r.connect()
conn.use('app')
/**
* Create socket server
*/
const wss = new WebSocket.Server({ port: 8082 })
wss.on('connection', (client) => {
/**
* Client setting subscribe rethinkdb data
* WARNING : doesn't validate payload
*/
client.on('message', (payload) => {
const filterData = JSON.parse(payload)
/**
* Send all data base on filter
*
/**
* Subscribe rethink data base on message request
* Send all data base on update
*/
r.table('foo')
.changes({
'includeTypes': true,
'includeInitial': true,
"squash": true // Squash many changes into one event
})
.run(conn, async (err, cursor) => {
if (err) {
throw err
}
client.cursor = cursor
cursor.each((err, row) => {
if (err) {
throw err
}
/**
* Check if client is connected
* Else close cursor
*
* Also useful for check if connect is dead
*/
if (client.readyState !== client.OPEN) {
client.terminate()
client.cursor.close()
return
}
client.send(JSON.stringify(row))
})
/**
* Close stream when :
* Client disconnect
* Client send new message
*/
client.on('message', () => {
cursor.close()
})
})
})
})
} catch (err) {
console.log(err)
}
}
start()
错误
ReqlDriverError: Cursor is closed.
at ReqlDriverError.ReqlError [as constructor] (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\errors.js:23:13)
at new ReqlDriverError (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\errors.js:68:50)
at Feed.IterableResult._promptNext (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:128:10)
at d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:200:22
at tryCatcher (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\util.js:26:23)
at Function.Promise.fromNode (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:168:30)
at Feed.<anonymous> (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:203:20)
at Feed.<anonymous> (d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\util.js:43:16)
at d:\project\vietnam-traffic-map\websocket\node_modules\rethinkdb\cursor.js:262:24
at tryCatcher (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\util.js:26:23)
at Promise.successAdapter (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\nodeify.js:23:30)
at Promise._settlePromiseAt (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:582:21)
at Promise._settlePromises (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\promise.js:700:14)
at Async._drainQueue (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:123:16)
at Async._drainQueues (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:133:10)
at Immediate.Async.drainQueues (d:\project\vietnam-traffic-map\websocket\node_modules\bluebird\js\main\async.js:15:14)
重现问题::
- 连接一个
- 重新连接(之前的连接已经关闭,但它的事件还没有关闭)
- 在rethinkdb中更改数据(此时我在if语句中打了个断点)
- 第一个连接关闭 -> 开始关闭游标并停止执行下一个
- 第二个还没关闭
- 错误:游标已关闭
问题: changefeeds 是否具有相同的过滤器也具有相同的光标?每个 websocket 客户端都应该有自己的 changefeed 游标。为什么我关闭了第一个的套接字,然后第二个也关闭了?
改变这个
if (client.readyState !== client.OPEN) {
client.terminate()
client.cursor.close()
return
}
client.send(JSON.stringify(row))
进入这个:
if (client.readyState === client.OPEN) {
client.send(JSON.stringify(row))
} else {
await cursor.close()
client.terminate()
}
解决我的问题。我仍然收到错误消息:ReqlDriverError: Cursor is closed. in try catch...