Manually polling streams in future implementation

Content Index :

Manually polling streams in future implementation
RSQM - polling manually?
Using polling with isDone and Cancel on Java Future instead of blocking get
Slight difference in Future.zip and Future.zipWith implementation. Why?
BizTalk manually fetch file instead of polling
Polling for specific time in the future

Manually polling streams in future implementation

Tag : development , By : DarrenBeck

Date : January 12 2021, 07:00 PM

wish helps you I'm in the process of migrating to futures 0.3 and tokio 0.2, and there is one recurring pattern I can't manage to re-use. I'm not sure whether this pattern became obsolete or whether I'm doing something wrong wrt to Pin. , You can use Pin::as_mut to avoid consuming the Pin.

impl MyFuture {
    fn poll_data(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
        use Poll::*;

        let MyFuture {
            ref mut data,
            ref mut state,
            ..
        } = self.get_mut();

        let mut data = Pin::new(data); // Move pin here
        loop {
            match data.as_mut().poll_next(cx) {   // Use in loop by calling `as_mut()`
                Ready(Some(vec)) => state.update(vec),
                Ready(None) => return Ready(()),
                Pending => return Pending,
            }
        }
    }
}

impl Future for MyFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        use Poll::*;
        // `as_mut()` here to avoid consuming
        if let Ready(_) = self.as_mut().poll_data(cx) { 
            return Ready(());
        }

        // can consume here as this is the last invocation
        if let Ready(_) = self.poll_events(cx) {
            return Ready(());
        }
        return Pending;
    }
}

impl MyFuture {
    fn poll_data(&mut self, cx: &mut Context) -> Poll<()> {
        use Poll::*;

        loop {
            match Pin::new(&mut self.data).poll_next(cx) {
                Ready(Some(vec)) => self.state.update(vec),
                Ready(None) => return Ready(()),
                Pending => return Pending,
            }
        }
    }
}

impl Future for MyFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
        use Poll::*;
        if let Ready(_) = self.poll_data(cx) {
            return Ready(());
        }

        if let Ready(_) = self.poll_events(cx) {
            return Ready(());
        }
        return Pending;
    }
}

Comments

No Comments Right Now !

Boards Message :

You Must Login Or Sign Up to Add Your Comments .

Share :

Tags :

Manually polling streams in future implementation , Manually , polling , streams , future , implementation ,

RSQM - polling manually?

Tag : javascript , By : user121501

Date : March 29 2020, 07:55 AM

To fix the issue you can do RSMQ itself does not implement any kind of worker, and ist stripped down to the core functionalities of a queue (which in my opinion is the correct approach). While you could implement everything yourself of course, I'd recommend using the additional module rsmq-worker (http://smrchy.github.io/rsmq/rsmq-worker/) which provides the basic worker skeleton for your application.
They provide a simple but usable example on their site:

  var RSMQWorker = require( "rsmq-worker" );
  var worker = new RSMQWorker( "myqueue" );

  worker.on( "message", function( msg, next ){
      // process your message
      next()
  });

  // optional error listeners
  worker.on('error', function( err, msg ){
      console.log( "ERROR", err, msg.id );
  });
  worker.on('exceeded', function( msg ){
      console.log( "EXCEEDED", msg.id );
  });
  worker.on('timeout', function( msg ){
      console.log( "TIMEOUT", msg.id, msg.rc );
  });

  worker.start();

Using polling with isDone and Cancel on Java Future instead of blocking get

Tag : java , By : Frank

Date : March 29 2020, 07:55 AM

should help you out Asynchronously polling a set of Futures using a separate thread does sound like a reasonable implementation to me. That said, if you're able to add a library dependency, you might find it easier to switch to Guava's ListenableFuture, as Guava provides a wealth of utilities for doing asynchronous work.

Slight difference in Future.zip and Future.zipWith implementation. Why?

Tag : scala , By : Genipro

Date : March 29 2020, 07:55 AM

Does that help The application of f is done with the supplied ExecutionContext, and internalExecutor is used to perform the flattening operation. The rule is basically: When the user supplies the logic, that logic is executed on the ExecutionContext supplied by the user.
You could imagine that zipWith was implemented as this.zip(that).map(f.tupled) or that zip was implemented as zipWith(Tuple2.apply)(internalExecutor).