Listeners are not removed upon unsubscribe()

[Faibk]

If topic.unsubscribe() is called without a parameter, it will not remove the callbacks, but instead unregister the whole topic at the rosconnection. This is fine as long as the topic object is destroyed after unsubscribing, but in my use case we subscribe to the same topic instance again. This adds a new callback and re-registers at the ros connection, which means now two callbacks are executed...

The problem seems to be acknowledged in this code comment:

roslibjs/src/core/Topic.js

Line 125 in 47336fa

// Note: Don't call this.removeAllListeners, allow client to handle that themselves

I do not see why the client would want to handle this itself, as it seems to be the expected outcome. unsubscribe(X) removes callback X from the topic's listener-list so the paradigm seems to be "unsubscribe X from the topic`. This should be the same without parameters, but currently it is "unsubscribe the topic from ros", which changes the direction of the method.

I propose not only unsubscribing to ros, but also removing all listener callbacks.

[jaguardo]

Curious if anyone else has experienced this or have a good work around?

[Rayman]

I'm not sure in what scenario to use unsubscribe(). I always use unsubscribe(x) with a callback.

[Faibk]

Essentially the scenario is if you register multiple subscribers on the topic and want to close all at once/shut down the whole topic (and later reconnect without a page refresh). Ideally one would always keep track of all registered callback listeners, but that might not be the case or there might be connection issues and you reload only parts of the state (we had modal views in one application, connecting and disconnecting on open/close).

From my point of view it is not about whether the feature is used, but what the expected behaviour is as long as it exists. It should either be removed or work like expected. Making a difference in behaviour beween unsubscribe(x) with argument and unsubscribe() without argument imho creates confusion. Especially since the behaviour seems to be undocumented, but needs to be handled by the client according to the code comment. It is also a memory leak if used without handling the removal of the listeners manually.

Regarding a workaround -- I use a fork with some modifications and just pull regularly new versions from the official repo.

[Rayman]

I think you could interpret a unsubscribe() call in two ways:

1. Remove all previously registered listeners
2. Stop listening to the topic, but keep the listeners in tact (essentially pausing the topic)

For 2 the problem is there is no way to resubscribe all the topics that were unsubscribed with a unsubscribe() call? I guess it's intended to be used like you said. You create a component that displays a video stream for example, and then the user switches to a different tab/modal and you want to "pause" the subscription. And later re-enable without fiddling with the listeners.

[jaguardo]

Perhaps I'm not using it the way it was intended either... but I create a listener

    listener = createListener(topic_path,
          topic_msgType,
          Number(queue),
          compression);
    }

and here is what it looks like:

listener: Topic
  callForSubscribeAndAdvertise: ƒ (message)
  compression: "none"
  isAdvertised: false
  latch: false
  messageType: "geometry_msgs/Twist"
  name: "/turtle1/cmd_vel"
  queue_length: 0
  queue_size: 100
  reconnect_on_close: true
  ros: Ros {socket: WebSocket, idCounter: 1, isConnected: true, transportLibrary: "websocket", transportOptions: {…}, …}
  throttle_rate: 0
  _messageCallback: ƒ (data)
  __proto__: EventEmitter

I subscribe to the message:

listener.subscribe(handleMsg);

and now it looks like this (note the subscribeID and _events.message:

listener: Topic
  callForSubscribeAndAdvertise: ƒ (message)
  compression: "none"
  isAdvertised: false
  latch: false
  messageType: "geometry_msgs/Twist"
  name: "/turtle1/cmd_vel"
  queue_length: 0
  queue_size: 100
  reconnectFunc: ƒ ()
  reconnect_on_close: true
  ros: Ros {socket: WebSocket, idCounter: 2, isConnected: true, transportLibrary: "websocket", transportOptions: {…}, …}
  subscribeId: "subscribe:/turtle1/cmd_vel:2"
  throttle_rate: 0
  waitForReconnect: false
  _events:
    message: ƒ handleMsg(msg)
    arguments: (...)
    caller: (...)
    length: 1
    name: "handleMsg"
    prototype: {constructor: ƒ}
    __proto__: ƒ ()
    [[FunctionLocation]]: RosConnect.jsx:407
    [[Scopes]]: Scopes[4]
  __proto__: Object
_messageCallback: ƒ (data)
__proto__: EventEmitter

regardless of how I try to unsubscribe:

      listener.unsubscribe();
      listener.unsubscribe(handleMsg);

the only thing that happens (at least that I can visibly see) is that subscribeId remains and becomes null... the callback is still there.

listener: Topic
  callForSubscribeAndAdvertise: ƒ (message)
  compression: "none"
  isAdvertised: false
  latch: false
  messageType: "geometry_msgs/Twist"
  name: "/turtle1/cmd_vel"
  queue_length: 0
  queue_size: 100
  reconnectFunc: ƒ ()
  reconnect_on_close: true
  ros: Ros {socket: WebSocket, idCounter: 2, isConnected: true, transportLibrary: "websocket", transportOptions: {…}, …}
  subscribeId: null
  throttle_rate: 0
  waitForReconnect: false
  _events:
    message: ƒ handleMsg(msg)
    arguments: (...)
    caller: (...)
    length: 1
    name: "handleMsg"
    prototype: {constructor: ƒ}
    __proto__: ƒ ()
    [[FunctionLocation]]: RosConnect.jsx:407
    [[Scopes]]: Scopes[4]
  __proto__: Object
_messageCallback: ƒ (data)
__proto__: EventEmitter

now when I try to re-subscribe I get 2 call backs instead of one:

listener: Topic
  callForSubscribeAndAdvertise: ƒ (message)
  compression: "none"
  isAdvertised: false
  latch: false
  messageType: "geometry_msgs/Twist"
  name: "/turtle1/cmd_vel"
  queue_length: 0
  queue_size: 100
  reconnectFunc: ƒ ()
  reconnect_on_close: true
  ros: Ros {socket: WebSocket, idCounter: 3, isConnected: true, transportLibrary: "websocket", transportOptions: {…}, …}
  subscribeId: "subscribe:/turtle1/cmd_vel:3"
  throttle_rate: 0
  waitForReconnect: false
  _events:
    message: Array(2)
      0: ƒ handleMsg(msg)
      1: ƒ handleMsg(msg)
      length: 2
      __proto__: Array(0)
    __proto__: Object
_messageCallback: ƒ (data)
__proto__: EventEmitter

and thus 2 messages instead of 1...

here is my workaround for now:

listener._events.message = null;

(but have to appropriately check that it is defined first)...

[Rayman]

Are you storing the handle so you can later unsubscribe?

var handle = listener.subscribe(handleMsg);
listener.unsubscribe(handle);

[jaguardo]

yes, I've tried it that way... though I believe the documentation (http://robotwebtools.org/jsdoc/roslibjs/current/Topic.html) implies that you use the handleMsg handle to unsubscribe.

Admittedly I'm instantiating the listener through react-ros (https://github.com/flynneva/react-ros), but it looks like a pass-through to roslib for the listener creation:

  function createListener(topic, msg_type, to_queue, compression_type) {
    var newListener = new ROSLIB.Topic({
      ros: ros.ROS,
      name: topic,
      messageType: msg_type,
      queue_length: to_queue,
      compression: compression_type
    });
    
    for (var listener in ros.listeners) {
      if (newListener.name === ros.listeners[listener].name) {
        console.log('Listener already available in ros.listeners[' + listener + ']');
        return ros.listeners[listener];
      }
    }

    ros.listeners.push(newListener);
    console.log('Listener created');
    return newListener;
  }

when I try to save the handle of the subscribe var handle = listener.subscribe(handleMsg);, it looks like it handle is just an undefined reference... so maybe that's the issue is that the handle isn't created correctly?

this issue (#165) seems to imply that unsubscribe does not touch the callback and instead need to use topic.removeAlllisteners?

same issue, still open here: #343

[jaguardo]

Can confirm that listener.removeAllListeners() does what is implied by listener.unsubscribe(handleMsg)
here is what listener looks like after that call:

listener: Topic
  callForSubscribeAndAdvertise: ƒ (message)
  compression: "none"
  event: "messaage"
  isAdvertised: false
  latch: false
  messageType: "geometry_msgs/Twist"
  name: "/turtle1/cmd_vel"
  queue_length: 0
  queue_size: 100
  reconnectFunc: ƒ ()
  reconnect_on_close: true
  ros: Ros {socket: WebSocket, idCounter: 2, isConnected: true, transportLibrary: "websocket", transportOptions: {…}, …}
  subscribeId: null
  throttle_rate: 0
  waitForReconnect: false
  _events:
    __proto__: Object
_messageCallback: ƒ (data)
__proto__: EventEmitter

I tried it with listener.removeAllListeners(handleMsg) and it didn't work the same/correctly, still got duplicate messages and it simply added a interesting quoted text to the listener:

  _events:
    "function handleMsg(msg) {↵ console.log(msg); // listener.unsubscribe();↵ }": null
    message: Array(2)
      0: ƒ handleMsg(msg)
      1: ƒ handleMsg(msg)
      length: 2
      __proto__: Array(0)
    __proto__: Object