Issue #1544 Timeout unacknowledged config management operation requests

[albinsuresh]

Proposed changes

If a child device does not respond to a config management operation request, fail that operation in the cloud with a timeout.
This PR is split into two parts:

[x] Make the plugin stateful so that the request-response mappings can be tracked
[] Timeout unacknowledged config management operation requests

Types of changes

• Bugfix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Improvement (general improvements like code refactoring that doesn't explicitly fix a bug or add any new functionality)
• Documentation Update (if none of the other choices apply)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)

Paste Link to the issue

#1544

Checklist

• I have read the CONTRIBUTING doc
• I have signed the CLA (in all commits with git commit -s)
• I ran cargo fmt as mentioned in CODING_GUIDELINES
• I used cargo clippy as mentioned in CODING_GUIDELINES
• I have added tests that prove my fix is effective or that my feature works
• I have added necessary documentation (if appropriate)

Further comments

[albinsuresh]

Unfortunately tokio::task::yield_now() didn't work here. Not quite sure why. So, resorted to tokio::time::sleep() instead.

[didier-wenzek]

Approved. However, code clarity can be improved by encapsulating the timer logic.

type TimerFuture<T> = Pin<Box<dyn Future<Output = T> + Send + Sync + 'static>>;

pub struct Timers<T> {
    pending_timers: FuturesUnordered<TimerFuture<T>>,
}

impl<T: Sync + Send + 'static> Timers<T> {
    pub fn new() -> Self {
        Timers {
            pending_timers: FuturesUnordered::new(),
        }
    }

    pub fn pending(&mut self) -> &mut FuturesUnordered<TimerFuture<T>> {
        &mut self.pending_timers
    }

    pub fn add_timer(&mut self, timeout: Duration, value: T) {
        self.pending_timers.push(Box::pin(sleep_and_return(timeout, value)));
    }
}

async fn sleep_and_return<T: Sync + Send + 'static>(timeout: Duration, value: T) -> T {
    sleep(timeout).await;
    value