mavlinkDirect: subscribe to several mavlink messages simultaneously

[Podrosboy]

How do I subscribe to several mavlink messages simultaneously if these messages are from custom xml?

I subscribe to the "OBSTACLE_DISTANCE" command and one custom command. If I subscribe to only one command, both work fine, but if I subscribe to both at the same time, only the first one works, and the subsequent one starts arriving simultaneously and doesn't transmit data.

drone.mavlinkDirect.getMessage("AGRO_TANK_STATUS")
    .subscribe({ mavlinkMessageAgroTankStatus ->
        val deserialized: AgroTankStatus =
            jsonAgroTankStatus.decodeFromString<AgroTankStatus>(
                mavlinkMessageAgroTankStatus.fieldsJson
            )
        println("AgroTankStatus: weight - ${deserialized.weight}")
    }, { error ->
        Log.e("MAVLink", "AgroTankStatus error", error)
    })


drone.mavlinkDirect.getMessage("OBSTACLE_DISTANCE")
    .subscribe { mavlinkMessage ->
        val deserialized: ObstacleDistance =
            jsonObstacleDistance.decodeFromString<ObstacleDistance>(
                mavlinkMessage.fieldsJson
            )
        println(
            "ObstacleDistance: distances - ${
                deserialized.distances.joinToString(
                    separator = ", "
                )
            }"
        )
    }

[JonasVautherin]

@julianoes: is that expected to work? Subscribing twice in parallel, with two different filters?

Looking at the sources, it seems like it should be supported...

@Podrosboy: could you try reproducing your issue in another language, e.g. using MAVSDK-Python or MAVSDK-C++? Just to see if that is a MAVSDK issue or something on your end. For instance you could try editing an example like this one.

[julianoes]

I think it should be supported yes, at least in C++.

[Podrosboy]

On C++ it is working. Can you please tell me if it's MAVSDK-Java problem or in my source code? What steps need to be taken to debug this behavior?

[JonasVautherin]

On C++ it is working.

Thanks for testing, good to know! So that part works (MAVSDK-Java runs the C++ code, so it's important).

Next, I would try with MAVSDK-Python. That would validate the gRPC layer. Now that I think about it, I am not completely sure if it works to subscribe to the same function twice in parallel 🤔. If you can reproduce the issue from MAVSDK-Python, it will hint towards an issue at the gRPC level.

EDIT: for what I can see in the gRPC implementation, I would say that it should work. I don't see a limitation there... so yeah, trying from Python would be valuable!

[Podrosboy]

Python is also working, so, no issue at the gRPC level. This is getting really mysterious...

[JonasVautherin]

So either it's in your code, or it's in the MAVSDK-Java implementation 😬.

Instead of calling drone.mavlinkDirect.getMessage in your code above, can you try the exact same structure but with a different observable? Say you call drone.telemetry.getPosition twice. Then you check if you receive events for both subscriptions or not?

The next thing would be to look at the implementation of drone.mavlinkDirect.getMessage. If you follow it in your IDE, does it show the implementation? Could you paste it here?

[Podrosboy]

I request the telemetry that can be obtained through drone.telemetry in the exact same way, and there are 15 of them in total, and they all work perfectly, the problem only arises with drone.mavlinkDirect

// Battery
    drone.telemetry.battery
        .subscribe({ battery ->
            updateDroneState {
                copy(
                    batteryStateParam = Battery(
                        temperatureDegc = battery.temperatureDegc,
                        voltageV = battery.voltageV,
                        currentBatteryA = battery.currentBatteryA,
                        capacityConsumedAh = battery.capacityConsumedAh,
                        remainingPercent = battery.remainingPercent,
                    ),
                )
            }
        }, { error ->
            Log.e("MAVLink", "Battery error", error)
        })
  
// StatusText
    drone.telemetry.statusText
        .subscribe({ statusText ->
            addErrorMessage(statusText)
        }, { error ->
            Log.e("MAVLink", "StatusText error", error)
        })

I tried running two drone.telemetry.battery subscriptions at the same time and everything worked.

Here is the implementation of the function mavlinkDirect.getMessage

@CheckReturnValue
  public Flowable<MavlinkDirect.MavlinkMessage> getMessage(@NonNull String messageName) {
    if (!isMessageInitialized) {
      synchronized (this) {
        if (!isMessageInitialized) {
          MavsdkEventQueue.executor().execute(() -> processMessage(messageName));
          isMessageInitialized = true;
        }
      }
    }
    return message;
  }

[JonasVautherin]

I tried running two drone.telemetry.battery subscriptions at the same time and everything worked.

Interesting 🤔

Here is the implementation of the function mavlinkDirect.getMessage

Could you show the implementation of processMessage(messageName)) as well? Or if you could just paste the whole file somewhere... I just don't have an IDE setup with MAVSDK-Java at the moment to go get the generated sources 😁

[Podrosboy]

Here is the whole file MavlinkDirect

mavlinkDirect.java

[JonasVautherin]

Oh, maybe I see it:

    private final FlowableProcessor<MavlinkDirect.MavlinkMessage> messageProcessor = PublishProcessor.create();
    private final Flowable<MavlinkDirect.MavlinkMessage> message = messageProcessor.onBackpressureBuffer().share();

    private volatile boolean isMessageInitialized = false;

    private void processMessage(@NonNull String messageName) {
        MavlinkDirectProto.SubscribeMessageRequest request = MavlinkDirectProto.SubscribeMessageRequest.newBuilder()
                .setMessageName(messageName)
                .build();

        stub.subscribeMessage(request, new StreamObserver<MavlinkDirectProto.MessageResponse>() {

            @Override
            public void onNext(MavlinkDirectProto.MessageResponse value) {
                messageProcessor.onNext(MavlinkMessage.translateFromRpc(value.getMessage()));
            }

            @Override
            public void onError(Throwable t) {
                messageProcessor.onError(t);
            }

            @Override
            public void onCompleted() {
                messageProcessor.onComplete();
            }
        });
    }

    @CheckReturnValue
    public Flowable<MavlinkDirect.MavlinkMessage> getMessage(@NonNull String messageName) {
        if (!isMessageInitialized) {
            synchronized (this) {
                if (!isMessageInitialized) {
                    MavsdkEventQueue.executor().execute(() -> processMessage(messageName));
                    isMessageInitialized = true;
                }
            }
        }
        return message;
    }

I think that we share one Flowable between all calls of getMessage. Which was necessary in the past because MAVSDK-C++ only supported exactly one subscription for each stream, but that is no longer the case. Also I guess it makes sense if there are no arguments to the stream, because then we can share it (like share the same Flow for the positions makes sense). But with an argument (here the messageName), this doesn't work anymore.

If I am correct, it means that when you register drone.mavlinkDirect.getMessage("OBSTACLE_DISTANCE") after drone.mavlinkDirect.getMessage("AGRO_TANK_STATUS"), you do receive events, but those are AGRO_TANK_STATUS events. Could you confirm that? I.e. that the mavlinkMessages you receive in the second subscription actually correspond to the first subscription?

[Podrosboy]

You are absolutely correct!
When I register drone.mavlinkDirect.getMessage("OBSTACLE_DISTANCE") after drone.mavlinkDirect.getMessage("AGRO_TANK_STATUS"), I do receive events and those are AGRO_TANK_STATUS events. Because the second one always had a different deserialization, it was outputting incorrect data and I thought it wasn't working at all.

[JonasVautherin]

I'll try to come up with a fix when I have time, maybe this weekend or the next 😊

[JonasVautherin]

Trying to fix it here. Again, I don't have a setup to test it so it would be appreciated if you could try it 😊.

You should be able to pull the SNAPSHOT following the instructions in the README. You want to try 3.14.0-1-SNAPSHOT.

Let me know how that goes! 🤞

For the record, I changed the implementation of the infinite streams to:

    @CheckReturnValue
    public Flowable<MavlinkDirect.MavlinkMessage> getMessage(@NonNull String messageName) {
      return Flowable.create(emitter -> {
        MavsdkEventQueue.executor().execute(() -> {
          MavlinkDirectProto.SubscribeMessageRequest request = MavlinkDirectProto.SubscribeMessageRequest.newBuilder()
            .setMessageName(messageName)
            .build();

          stub.subscribeMessage(request, new StreamObserver<MavlinkDirectProto.MessageResponse>() {

            @Override
            public void onNext(MavlinkDirectProto.MessageResponse value) {
              emitter.onNext(MavlinkMessage.translateFromRpc(value.getMessage()));
            }

            @Override
            public void onError(Throwable t) {
              emitter.onError(t);
            }

            @Override
            public void onCompleted() {
              emitter.onComplete();
            }
          });
        });
      }, BackpressureStrategy.LATEST);
    }

So that the stream shouldn't be shared anymore, and it should not cause the issue above.

[Podrosboy]

Yes, that helped, everything works now, thanks.
Will this change be in the next release?