Skip to content

Types issues #253

Open
Open
Types issues#253
@shovelmn12

Description

@shovelmn12
shovelmn12
opened on Nov 29, 2024

Hello

I am working on a small project to create a backend framework to handle events (microservices)
I am trying to implement the pipe to handle the events but I keep getting errors when calling subscribe()

Generally I dont get why is the implementation working this way.

  1. Why dont extend Stream?
  2. Why not return a unified definition of Observable? Each OP returns a different defenition therefore creating a mess when trying to read errors (see below)

For reference check rxdart. They started with their own definition then went back to create operation extension on Stream from dart to solve all those issues, also each Op returns Stream and not MapOp, FilterOp etc ...

The code:

Simple version

    from_future_result(ready(Ok("")), threads_scheduler)
        .flat_map(|_| from_future_result(ready(Ok("")), threads_scheduler.clone()))
        .on_error(|error| {})
        .subscribe(|e| {}) // error here

Full version

#[derive(Clone)]
pub struct EventSource<Context, Event, Error>
where
    Context: IntoBody + Sized + Send + Sync + 'static,
    Event: IntoBody + Sized + Send + Sync + 'static,
    Error: IntoBody + Sized + Send + Sync + 'static,
{
    pub tx: Subject<'static, (Context, Event), Error>,
    pub rx: Subject<'static, (Context, Event), Error>,
}

pub type EventHandler<Context, Event, Error> =
    fn((Context, Event)) -> Pin<Box<dyn Future<Output = Result<(), Error>>>>;
pub type CreateSource<Context, Event, Error> =
    fn() -> Pin<Box<dyn Future<Output = Result<EventSource<Context, Event, Error>, Error>>>>;

pub fn start_service<Context, Event, Error>(
    source: CreateSource<Context, Event, Error>,
    handler: EventHandler<Context, Event, Error>,
) -> Box<dyn Subscription>
where
    Context: IntoBody + Sized + Sync + Send,
    Event: IntoBody + Sized + Sync + Send,
    Error: IntoBody + Sized + Sync + Send,
{
    let threads_scheduler = futures::executor::ThreadPool::new().unwrap();

    Box::new(
        from_future_result(source(), threads_scheduler.clone())
            .flat_map_threads(|source| source.rx)
            .flat_map_threads(|event| from_future_result(handler(event), threads_scheduler.clone()))
            .on_error(|error| {})
            .subscribe(|result| {}), // issue is here
    )
}

Full error:

Details 

    the method `subscribe` exists for struct `OnErrorOp<MergeAllOpThreads<MapOp<MergeAllOpThreads<MapOp<..., ..., ...>, ...>, ..., ...>, ...>, ..., ...>`, but its trait bounds were not satisfied [E0599] Note: the full type name has been written to '/ Users/ shovelmn12/ Projects/ poc_server/ target/ debug/ deps/ poc_server-b47a2d7829a631fd. long-type-1706698521457854592.txt' Note: consider using `--verbose` to print the full type name to the console Note: the following trait bounds were not satisfied: `rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>` `&rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `&rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>` `&mut rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `&mut rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>`
    the method `subscribe` exists for struct `OnErrorOp<MergeAllOpThreads<MapOp<MergeAllOpThreads<MapOp<..., ..., ...>, ...>, ..., ...>, ...>, ..., ...>`, but its trait bounds were not satisfied [E0599] Note: the full type name has been written to '/ Users/ shovelmn12/ Projects/ poc_server/ target/ debug/ deps/ poc_server-deb9ac863f34f807.long-type-7867709162962570806.txt' Note: consider using `--verbose` to print the full type name to the console Note: the following trait bounds were not satisfied: `rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>` `&rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `&rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>` `&mut rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::Observable<_, std::convert::Infallible, rxrust::observable::ObserverItem<_>>` which is required by `&mut rxrust::ops::on_error::OnErrorOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::ops::merge_all::MergeAllOpThreads<rxrust::ops::map::MapOp<rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<EventSource<Context, Event, Error>, Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>, {closure@src/ main. rs:54:31: 54:39}, EventSource<Context, Event, Error>>, rxrust::subject::Subject<'static, (Context, Event), Error>>, {closure@src/ main. rs:55:31: 55:38}, (Context, Event)>, rxrust::observable::from_future::FutureResultObservable<std::pin::Pin<std::boxed::Box<dyn futures::Future<Output = std::result::Result<(), Error>>>>, rxrust::scheduler::FuturesThreadPoolScheduler>>, {closure@src/ main. rs:56:23: 56:30}, Error>: rxrust::observable::ObservableItem<_, _>`

FYI: not a rust expert, just honestly trying to understand :)

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions