1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
// Copyright 2019 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

use crate::hanging_get::error::HangingGetServerError;
use core::hash::Hash;
use fuchsia_sync::Mutex;
use std::collections::HashMap;
use std::sync::Arc;

/// A broker of updates to some state, using the hanging get pattern.
///
/// The broker stores a state, and mediates state changes between
/// [Publisher]s and [Subscriber]s.  Any [Publisher] can change the state at any
/// time.  When the state is changed, all [Subscriber]s are notified with the new
/// state.
///
/// To follow the hanging get pattern, when a [Subscriber] is first created,
/// the current state is sent immediately.  Subsequent updates to the same
/// subscriber are only sent when the state is modified.
///
/// * Use [HangingGet::new] to create a new broker for a single state item.
/// * Use [HangingGet::new_publisher] to create an object that allows you to
///   update the state.
/// * Use [HangingGet::new_subscriber] to create an object that monitors for
///   updates to the state.
///
/// ## Type parameters
///
/// * `S`: the type of the stored hanging get state.  This is the state that gets
///   communicated to subscribers.
/// * `O`: The type of the object used to notify of state change.
/// * `F`: The type of a function used to send the new state content to an instance
///   of `O`.  `F` gets passed the content of the new state, the object that it
///   needs to notify, and is expected to return `true` if the notification was
///   a success; otherwise it must return `false`.
pub struct HangingGet<S, O, F: Fn(&S, O) -> bool> {
    inner: Arc<Mutex<HangingGetInner<S, subscriber_key::Key, O, F>>>,
    /// A [subscriber_key::Key] held by the broker to track the next unique key that the broker can
    /// assign to a [Subscriber].
    subscriber_key_generator: subscriber_key::Generator,
}

impl<S, O, F> HangingGet<S, O, F>
where
    F: Fn(&S, O) -> bool,
{
    /// Create a new broker.
    ///
    /// ## Args:
    ///
    /// * `state` is the initial state of the [HangingGet].
    /// * `notify` is a function to notify observers of state of the state change.
    pub fn new(state: S, notify: F) -> Self {
        Self {
            inner: Arc::new(Mutex::new(HangingGetInner::new(Some(state), notify))),
            subscriber_key_generator: subscriber_key::Generator::default(),
        }
    }

    /// Create a new broker, but delays any subscribers from being notified until the value is
    /// initialized.
    ///
    /// ## Args:
    ///
    /// * `notify` is a function to notify observers of state of the state change.
    ///
    /// # Disclaimer:
    /// This initialier is more prone to cause hangs if code is not properly setup. This
    /// should only be used for patterns where the is no useful default state.
    pub fn new_unknown_state(notify: F) -> Self {
        Self {
            inner: Arc::new(Mutex::new(HangingGetInner::new(None, notify))),
            subscriber_key_generator: subscriber_key::Generator::default(),
        }
    }

    /// Create a new [Publisher] that can make atomic updates to the state value.
    pub fn new_publisher(&self) -> Publisher<S, O, F> {
        Publisher { inner: self.inner.clone() }
    }

    /// Create a new [Subscriber] that represents a single hanging get client.
    ///
    /// The newly-created subscriber will be notified with the current state
    /// immediately.  After the first notification, the subscriber will be
    /// notified only if the state changes.
    pub fn new_subscriber(&mut self) -> Subscriber<S, O, F> {
        Subscriber { inner: self.inner.clone(), key: self.subscriber_key_generator.next().unwrap() }
    }
}

/// A `Subscriber` can be used to register observation requests with the `HangingGet`.
/// These will be notified when the state changes or immediately the first time
/// a `Subscriber` registers an observation.
///
/// ## Type parameters
///
/// See [HangingGet] for the explanation of the type parameters `S`, `O`, `F`.
pub struct Subscriber<S, O, F: Fn(&S, O) -> bool> {
    inner: Arc<Mutex<HangingGetInner<S, subscriber_key::Key, O, F>>>,
    key: subscriber_key::Key,
}

impl<S, O, F> Subscriber<S, O, F>
where
    F: Fn(&S, O) -> bool,
{
    /// Register a new observation.
    ///
    /// Errors occur when:
    /// * A Subscriber attempts to register an observation when there is already an outstanding
    ///   observation waiting on updates.
    pub fn register(&self, observation: O) -> Result<(), HangingGetServerError> {
        self.inner.lock().subscribe(self.key, observation)
    }
}

/// A `Publisher` is used to make changes to the state contained within a `HangingGet`.
/// It is designed to be cheaply cloned and `Send`.
///
/// ## Type parameters
///
/// See [HangingGet] for the explanation of the generic types `S`, `O`, `F`.
pub struct Publisher<S, O, F: Fn(&S, O) -> bool> {
    inner: Arc<Mutex<HangingGetInner<S, subscriber_key::Key, O, F>>>,
}

impl<S, O, F: Fn(&S, O) -> bool> Clone for Publisher<S, O, F> {
    /// Clones this [Publisher].
    ///
    /// It is cheap to clone a [Publisher].
    fn clone(&self) -> Self {
        Self { inner: self.inner.clone() }
    }
}

impl<S, O, F> Publisher<S, O, F>
where
    F: Fn(&S, O) -> bool,
{
    /// Set the stored state to `S`. Subscribers will be updated.
    pub fn set(&self, state: S) {
        self.inner.lock().set(state)
    }

    /// Pass a function to the hanging get that can update the hanging get state in place.
    ///
    /// Any subscriber that has registered an observer will immediately be notified of the
    /// update.
    ///
    /// ## Type parameters
    ///
    /// * `UpdateFn`: an update function: gets passed the new state, and returns `true`
    ///   if the state has been updated with success.
    pub fn update<UpdateFn>(&self, update: UpdateFn)
    where
        UpdateFn: FnOnce(&mut Option<S>) -> bool,
    {
        self.inner.lock().update(update)
    }
}

/// A [HangingGetInner] object manages some internal state `S` and notifies observers of type `O`
/// when their view of the state is outdated.
///
/// While it _can_ be used directly, most API consumers will want to use the higher level
/// [HangingGet] object. `HangingGet` and its companion types provide `Send`
/// for use from multiple threads or async tasks.
///
/// ## Type parameters
///
/// * `K` - the Key by which Observers are identified.
/// * For other type args see [HangingGet].
pub struct HangingGetInner<S, K, O, F: Fn(&S, O) -> bool> {
    state: Option<S>,
    notify: F,
    observers: HashMap<K, Window<O>>,
}

impl<S, K, O, F> HangingGetInner<S, K, O, F>
where
    K: Eq + Hash,
    F: Fn(&S, O) -> bool,
{
    fn notify_all(&mut self) {
        for window in self.observers.values_mut() {
            window.notify(&self.notify, self.state.as_ref().unwrap());
        }
    }

    /// Create a new `HangingGetInner`.
    /// `state` is the initial state of the HangingGetInner
    /// `notify` is a `Fn` that is used to notify observers of state.
    pub fn new(state: Option<S>, notify: F) -> Self {
        Self { state, notify, observers: HashMap::new() }
    }

    /// Set the internal state value to `state` and notify all observers.
    /// Note that notification is not conditional on the _value_ set by calling the `set` function.
    /// Notification will occur regardless of whether the `state` value differs from the value
    /// currently stored by `HangingGetInner`.
    pub fn set(&mut self, state: S) {
        self.state = Some(state);
        self.notify_all();
    }

    /// Modify the internal state in-place using the `state_update` function. Notify all
    /// observers if `state_update` returns true.
    pub fn update(&mut self, state_update: impl FnOnce(&mut Option<S>) -> bool) {
        if state_update(&mut self.state) {
            self.notify_all();
        }
    }

    /// Register an observer as a subscriber of the state.
    ///
    /// Observers are grouped by key and all observers will the same key are assumed to have
    /// received the latest state update. If an observer with a previously unseen key subscribes,
    /// it is immediately notified to the stated. If an observer with a known key subscribes, it
    /// will only be notified when the state is updated since last sent to an observer with the same
    /// key. All unresolved observers will be resolved to the same value immediately after the state
    /// is updated. If there is no stored state, then the notification will be delayed until an
    /// update is made.
    pub fn subscribe(&mut self, key: K, observer: O) -> Result<(), HangingGetServerError> {
        let entry = self.observers.entry(key).or_insert_with(Window::new);
        entry.observe(observer, &self.notify, self.state.as_ref())
    }

    /// Deregister all observers that subscribed with `key`. If an observer is subsequently
    /// subscribed with the same `key` value, it will be treated as a previously unseen `key`.
    pub fn unsubscribe(&mut self, key: K) {
        drop(self.observers.remove(&key));
    }
}

/// Window tracks all observers for a given `key` and whether their view of the state is
/// `dirty` or not.
struct Window<O> {
    dirty: bool,
    observer: Option<O>,
}

impl<O> Window<O> {
    /// Create a new `Window` without an `observer` and an initial `dirty` value of `true`.
    pub fn new() -> Self {
        Window { dirty: true, observer: None }
    }

    /// Register a new observer. The observer will be notified immediately if the `Window`
    /// has a dirty view of the state. The observer will be stored for future notification
    /// if the `Window` does not have a dirty view.
    pub fn observe<S>(
        &mut self,
        observer: O,
        f: impl Fn(&S, O) -> bool,
        current_state: Option<&S>,
    ) -> Result<(), HangingGetServerError> {
        if self.observer.is_some() {
            return Err(HangingGetServerError::MultipleObservers);
        }
        self.observer = Some(observer);
        if let Some(current_state) = current_state {
            if self.dirty {
                self.notify(f, current_state);
            }
        }
        Ok(())
    }

    /// Notify the observer _if and only if_ the `Window` has a dirty view of `state`.
    /// If an observer was present and notified, the `Window` no longer has a dirty view
    /// after this method returns.
    pub fn notify<S>(&mut self, f: impl Fn(&S, O) -> bool, state: &S) {
        match self.observer.take() {
            Some(observer) => {
                if f(state, observer) {
                    self.dirty = false;
                }
            }
            None => self.dirty = true,
        }
    }
}

/// Submodule used to keep the internals of `Key`s and `Generator`s inaccessable.
mod subscriber_key {
    /// Manages the creation and distribution of unique `Key`s
    pub struct Generator {
        next: Key,
    }

    impl Default for Generator {
        fn default() -> Self {
            Self { next: Key(0) }
        }
    }

    impl Generator {
        /// Get a unique key.
        /// Returns `None` if 2^64-2 keys have already been generated.
        pub fn next(&mut self) -> Option<Key> {
            let key = self.next.clone();
            if let Some(next) = self.next.0.checked_add(1) {
                self.next.0 = next;
                Some(key)
            } else {
                None
            }
        }
    }

    /// An internal per-subscriber key that is intended to be unique.
    #[derive(PartialEq, Eq, Hash, Debug, Clone, Copy)]
    pub struct Key(u64);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::hanging_get::test_util::TestObserver;
    use crate::PollExt;
    use fuchsia_async as fasync;
    use futures::channel::oneshot;

    #[test]
    fn subscriber_key_generator_creates_unique_keys() {
        let mut gen = subscriber_key::Generator::default();
        let key1 = gen.next();
        let key2 = gen.next();
        assert!(key1 != key2);
    }

    #[test]
    fn window_add_first_observer_notifies() {
        let state = 0;
        let mut window = Window::new();
        window
            .observe(TestObserver::expect_value(state), TestObserver::observe, Some(&state))
            .unwrap();
    }

    #[test]
    fn window_none_state_does_not_notify() {
        let mut window = Window::new();
        window
            .observe::<i32>(TestObserver::expect_no_value(), TestObserver::observe, None)
            .unwrap();
    }

    #[test]
    fn window_add_second_observer_does_not_notify() {
        let state = 0;
        let mut window = Window::new();
        window
            .observe(TestObserver::expect_value(state), TestObserver::observe, Some(&state))
            .unwrap();

        // Second observer added without updating the value
        window
            .observe(TestObserver::expect_no_value(), TestObserver::observe, Some(&state))
            .unwrap();
    }

    #[test]
    fn window_add_second_observer_notifies_after_notify_call() {
        let mut state = 0;
        let mut window = Window::new();
        window
            .observe(TestObserver::expect_value(state), TestObserver::observe, Some(&state))
            .unwrap();

        state = 1;
        window.notify(TestObserver::observe, &state);

        // Second observer added without updating the value
        window
            .observe(TestObserver::expect_value(state), TestObserver::observe, Some(&state))
            .unwrap();
    }

    #[test]
    fn window_add_multiple_observers_are_notified() {
        let mut state = 0;
        let mut window = Window::new();
        window
            .observe(TestObserver::expect_value(state), TestObserver::observe, Some(&state))
            .unwrap();

        // Second observer added without updating the value
        let o1 = TestObserver::expect_value(1);
        let o2 = TestObserver::expect_no_value();
        window.observe(o1.clone(), TestObserver::observe, Some(&state)).unwrap();
        let result = window.observe(o2.clone(), TestObserver::observe, Some(&state));
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        assert!(!o1.has_value());
        state = 1;
        window.notify(TestObserver::observe, &state);
    }

    #[test]
    fn window_dirty_flag_state() {
        let state = 0;
        let mut window = Window::new();
        let o = TestObserver::expect_value(state);
        window.observe(o, TestObserver::observe, Some(&state)).unwrap();
        assert!(window.observer.is_none());
        assert!(!window.dirty);
        window.notify(TestObserver::observe, &state);
        assert!(window.dirty);
        let o = TestObserver::expect_value(state);
        window.observe(o, TestObserver::observe, Some(&state)).unwrap();
        assert!(!window.dirty);
    }

    #[test]
    fn window_dirty_flag_respects_consumed_flag() {
        let state = 0;
        let mut window = Window::new();

        let o = TestObserver::expect_value(state);
        window.observe(o, TestObserver::observe_incomplete, Some(&state)).unwrap();
        assert!(window.dirty);
    }

    #[test]
    fn hanging_get_inner_subscribe() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        let o = TestObserver::expect_value(0);
        assert!(!o.has_value());
        hanging.subscribe(0, o.clone()).unwrap();
    }

    #[test]
    fn hanging_get_inner_subscribe_then_set() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        let o = TestObserver::expect_value(0);
        hanging.subscribe(0, o.clone()).unwrap();

        // No change without a new subscription
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_subscribe_twice_then_set() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(0)).unwrap();

        hanging.subscribe(0, TestObserver::expect_value(1)).unwrap();
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_subscribe_multiple_then_set() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(0)).unwrap();

        // A second subscription with the same client key should not be notified
        let o2 = TestObserver::expect_value(1);
        hanging.subscribe(0, o2.clone()).unwrap();
        assert!(!o2.has_value());

        // A third subscription will queue up along the other waiting observer
        let _ = hanging.subscribe(0, TestObserver::expect_no_value()).unwrap_err();

        // Set should notify all observers to the change
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_subscribe_with_two_clients_then_set() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(0)).unwrap();
        hanging.subscribe(0, TestObserver::expect_value(1)).unwrap();
        hanging.subscribe(1, TestObserver::expect_value(0)).unwrap();
        hanging.subscribe(1, TestObserver::expect_value(1)).unwrap();
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_unsubscribe() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(0)).unwrap();
        hanging.subscribe(0, TestObserver::expect_no_value()).unwrap();
        hanging.unsubscribe(0);
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_unsubscribe_one_of_many() {
        let mut hanging = HangingGetInner::new(Some(0), TestObserver::observe);

        hanging.subscribe(0, TestObserver::expect_value(0)).unwrap();
        hanging.subscribe(0, TestObserver::expect_no_value()).unwrap();
        hanging.subscribe(1, TestObserver::expect_value(0)).unwrap();
        hanging.subscribe(1, TestObserver::expect_no_value()).unwrap();

        // Unsubscribe one of the two observers
        hanging.unsubscribe(0);
        assert!(!hanging.observers.contains_key(&0));
        assert!(hanging.observers.contains_key(&1));
    }

    #[test]
    fn hanging_get_inner_delayed_subscribe() {
        let mut hanging = HangingGetInner::new(None, TestObserver::<u8>::observe);
        let o = TestObserver::expect_no_value();
        assert!(!o.has_value());
        hanging.subscribe(0, o.clone()).unwrap();
    }

    #[test]
    fn hanging_get_inner_delayed_subscribe_then_set() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);
        let o = TestObserver::expect_value(1);
        hanging.subscribe(0, o.clone()).unwrap();

        // Initial value now set.
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_delayed_subscribe_twice_then_set() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(1)).unwrap();

        // Since the first result is delayed, subscribing again is an error.
        let result = hanging.subscribe(0, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_delayed_subscribe_multiple_then_set() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(1)).unwrap();

        // A second subscription with the same client key should fail while the subscription hasn't
        // been notified
        let o2 = TestObserver::expect_no_value();
        let result = hanging.subscribe(0, o2.clone());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        assert!(!o2.has_value());

        // A third subscription will also fail for the same reason.
        let result = hanging.subscribe(0, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));

        // Set should notify all observers to the change
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_delayed_subscribe_with_two_clients_then_set() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_value(1)).unwrap();
        let result = hanging.subscribe(0, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        hanging.subscribe(1, TestObserver::expect_value(1)).unwrap();
        let result = hanging.subscribe(1, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_delayed_unsubscribe() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);
        hanging.subscribe(0, TestObserver::expect_no_value()).unwrap();
        let result = hanging.subscribe(0, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        hanging.unsubscribe(0);
        hanging.set(1);
    }

    #[test]
    fn hanging_get_inner_delayed_unsubscribe_one_of_many() {
        let mut hanging = HangingGetInner::new(None, TestObserver::observe);

        hanging.subscribe(0, TestObserver::<i32>::expect_no_value()).unwrap();
        let result = hanging.subscribe(0, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));
        hanging.subscribe(1, TestObserver::expect_no_value()).unwrap();
        let result = hanging.subscribe(1, TestObserver::expect_no_value());
        assert_eq!(result, Err(HangingGetServerError::MultipleObservers));

        // Unsubscribe one of the two observers
        hanging.unsubscribe(0);
        assert!(!hanging.observers.contains_key(&0));
        assert!(hanging.observers.contains_key(&1));
    }

    #[test]
    fn sync_pub_sub_updates_and_observes() {
        let mut ex = fasync::TestExecutor::new();
        let mut broker = HangingGet::new(0i32, |s, o: oneshot::Sender<_>| {
            o.send(s.clone()).map(|()| true).unwrap()
        });
        let publisher = broker.new_publisher();
        let subscriber = broker.new_subscriber();

        // Initial observation is immediate
        let (sender, mut receiver) = oneshot::channel();
        subscriber.register(sender).unwrap();
        let observation =
            ex.run_until_stalled(&mut receiver).expect("received initial observation");
        assert_eq!(observation, Ok(0));

        // Subsequent observations do not happen until after an update
        let (sender, mut receiver) = oneshot::channel();
        subscriber.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut receiver).is_pending());

        publisher.set(1);

        let observation =
            ex.run_until_stalled(&mut receiver).expect("received subsequent observation");
        assert_eq!(observation, Ok(1));
    }

    #[test]
    fn sync_pub_sub_multiple_subscribers() {
        let mut ex = fasync::TestExecutor::new();
        let mut broker = HangingGet::new(0i32, |s, o: oneshot::Sender<_>| {
            o.send(s.clone()).map(|()| true).unwrap()
        });
        let publisher = broker.new_publisher();

        let sub1 = broker.new_subscriber();
        let sub2 = broker.new_subscriber();

        // Initial observation for subscribers is immediate
        let (sender, mut receiver) = oneshot::channel();
        sub1.register(sender).unwrap();
        let observation =
            ex.run_until_stalled(&mut receiver).expect("received initial observation");
        assert_eq!(observation, Ok(0));

        let (sender, mut receiver) = oneshot::channel();
        sub2.register(sender).unwrap();
        let observation =
            ex.run_until_stalled(&mut receiver).expect("received initial observation");
        assert_eq!(observation, Ok(0));

        // Subsequent observations do not happen until after an update
        let (sender, mut recv1) = oneshot::channel();
        sub1.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut recv1).is_pending());

        let (sender, mut recv2) = oneshot::channel();
        sub2.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut recv2).is_pending());

        publisher.set(1);
        let obs1 =
            ex.run_until_stalled(&mut recv1).expect("receiver 1 received subsequent observation");
        assert_eq!(obs1, Ok(1));
        let obs2 =
            ex.run_until_stalled(&mut recv2).expect("receiver 2 received subsequent observation");
        assert_eq!(obs2, Ok(1));
    }

    #[test]
    fn sync_pub_sub_delayed_updates_and_observes() {
        let mut ex = fasync::TestExecutor::new();
        let mut broker = HangingGet::<i32, _, _>::new_unknown_state(|s, o: oneshot::Sender<_>| {
            o.send(s.clone()).map(|()| true).unwrap()
        });
        let publisher = broker.new_publisher();
        let subscriber = broker.new_subscriber();

        // Initial observation is delayed due to lack of initial state.
        let (sender, mut recv1) = oneshot::channel();
        subscriber.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut recv1).is_pending());

        // Subsequent registration fails since the original registration is still pending its
        // observation.
        let (sender, mut receiver) = oneshot::channel();
        assert!(subscriber.register(sender).is_err());
        assert!(ex.run_until_stalled(&mut receiver).expect("sender closed").is_err());

        // Initial observation received now that the initial value is set.
        publisher.set(1);

        let observation =
            ex.run_until_stalled(&mut recv1).expect("received subsequent observation");
        assert_eq!(observation, Ok(1));
    }

    #[test]
    fn sync_pub_sub_delayed_multiple_subscribers() {
        let mut ex = fasync::TestExecutor::new();
        let mut broker = HangingGet::<i32, _, _>::new_unknown_state(|s, o: oneshot::Sender<_>| {
            o.send(s.clone()).map(|()| true).unwrap()
        });
        let publisher = broker.new_publisher();

        let sub1 = broker.new_subscriber();
        let sub2 = broker.new_subscriber();

        // Initial observation for subscribers delayed due to lack of initial state.
        let (sender, mut recv1) = oneshot::channel();
        sub1.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut recv1).is_pending());

        let (sender, mut recv2) = oneshot::channel();
        sub2.register(sender).unwrap();
        assert!(ex.run_until_stalled(&mut recv2).is_pending());

        // Subsequent registrations fails since the original registrations are still pending their
        // observations.
        let (sender, mut recv3) = oneshot::channel();
        assert!(sub1.register(sender).is_err());
        assert!(ex.run_until_stalled(&mut recv3).expect("sender 3 closed").is_err());

        let (sender, mut recv4) = oneshot::channel();
        assert!(sub2.register(sender).is_err());
        assert!(ex.run_until_stalled(&mut recv4).expect("sender 4 closed").is_err());

        // Initial observations received now that initial value is set.
        publisher.set(1);
        let obs1 =
            ex.run_until_stalled(&mut recv1).expect("receiver 1 received subsequent observation");
        assert_eq!(obs1, Ok(1));
        let obs2 =
            ex.run_until_stalled(&mut recv2).expect("receiver 2 received subsequent observation");
        assert_eq!(obs2, Ok(1));
    }
}