Filter Events with subxt over multiple threads

Question

I'm working on a project that uses Substrate in the backend. The System I am working on is written with Elixir and so I have a rustler NIF (Erlang/Elixir Native Function interface) for interacting with subxt and our substrate client. A couple days ago, we bumped our subxt version up to the current dev branch. I am running into major issues trying to use the filter_events() event filter though because of the nature of how the Rustler NIF needs to work.

As a basic rundown, Rustler allows you to define native references and pass them back and forth between the Erlang Virtual Machine. In the case of a event listener, we are talking about a long running loop that iterates over the stream of events and passes them back in structs that can be handled by an associated Elixir Process. The issue with the new way of dealing with the Event's stream comes from the fact that some of the new structs don't implement the Send trait. More specifically: FilteredEventDetails and Events. Also, on the other side, the Env/Owned_Env for Rustler also doesn't implement Send which makes this tricky.

In the original code, I would pass in the client as a Elixir Term and decode it into the client type. It would reveal the runtime and use the RPC to subscribe to the events. The listener would match on each of the pallet and variant strings that matched the events that were needed and pass them directly back to a elixir process using an owned environment.

Here is a snippet:

let client: ResourceArc<ClientResource> = client.decode().expect("invalid client");
let pid = env.pid();

if let Some(client) = (*client).0.clone() {
    // Alias for tokio::task::spawn on the global runtime.
    let handle = spawn(async move {
        let client = client.clone();
        let runtime = client.runtime().client.clone();
        // Elixir Owned_env
        let mut owned_env = OwnedEnv::new();

        let sub = runtime
            .rpc()
            .subscribe_events()
            .await
            .expect("unable to subscribe");
        // event decoder and subscription
        let decoder = runtime.events_decoder();
        let mut sub = EventSubscription::new(sub, decoder);
        // iterate through all of the events
        while let Some(result) = sub.next().fuse().await {

            if let Ok(res) = result {
                match (&res.pallet, &res.variant) {
                    (p, v) if p == "Pallet" && v == "Variant" => {
                        // decode the Data event. 
                        let decoded = Data::decode(&mut &res.data[..]);
                        ...
                        match decoded {
                          Ok(Data { field_0, field_1, .... }) => {
                               // Send the data event back to Elixir in an ExData struct.
                               owned_env.send_and_clear(&pid, |e| {
                                  crate::atoms::data(),
                                  ExData { field_0, field_1, ...}
                                });
                             }
                        }
   ....

The outer functions are not async and the system uses a global lazy tokio runtime to deal with futures and streams.

static RUNTIME: Lazy<Runtime> = Lazy::new(|| Runtime::new().expect("failed to create runtime"));

Here's what the new code looks like in comparison:

let handle = spawn(async move {
    ...
    let runtime = client.runtime();
    let events = runtime.events();
    // First `Send` error is here.
    let sub = events.subscribe().await.expect("failed to subscribe");
    let events = sub.filter_events::<(Event_1, Event_2, Event_3, ...)>();
    // Second `Send` error from iterator (next) call.
    while let Some(res) = events.next().await {
       if let Ok(result) = res {
          match result.event {
            (Some(Event_1 { field_1, field_2, ... }, _, _, _, ...)) => {
              owned_env.send_and_clear(&pid, |e| {
                crate::atoms::event_1(),
                ExEvent_1 { field_1, field_2, ... }  
              });
            }
          }
       } else {
       ....
       }
    }

});

I've tried a few ways to attempt to squash these Send errors; I've attempted to use pin_mut! and Box::pin() on the two types, I've tried implementing Wrapper types that do implement Send, I've tried using LocalSet and tokio::task::spawn_local but none of these seem to work as solutions for this particular issue. My latest idea, saw me building the client in the spawned task with the listener but it also saw failure. Does anyone have any ideas of how I can deal with this issue? I've looked at all of the subxt examples, and most if not all of them do feature iterators but also are inside of a local thread via async-std::test or async-std::main neither of which I can use for this particular API. I know its probably possible for me to do some kind of trickery with channels where the messages are sent back to some other callback function but I want to avoid this if possible as its not very compatible with the Rustler Idioms.

Answer 1

Subxt v0.19.0 has now been released, which includes https://github.com/paritytech/subxt/pull/471. This makes the relevant types Send, which hopefully resolves your issue :)