Carlos Martinez Romero | 43d9afc | 2023-07-07 22:58:51 +0000 | [diff] [blame] | 1 | // Copyright (C) 2023 The Android Open Source Project |
| 2 | // |
| 3 | // Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | // you may not use this file except in compliance with the License. |
| 5 | // You may obtain a copy of the License at |
| 6 | // |
| 7 | // http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | // |
| 9 | // Unless required by applicable law or agreed to in writing, software |
| 10 | // distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | // See the License for the specific language governing permissions and |
| 13 | // limitations under the License. |
| 14 | |
| 15 | //! libbufferstreams: Reactive Streams for Graphics Buffers |
| 16 | |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 17 | pub mod publishers; |
James Shargo | 83baba5 | 2023-09-11 20:17:40 -0400 | [diff] [blame] | 18 | mod stream_config; |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 19 | pub mod subscribers; |
| 20 | pub mod subscriptions; |
James Shargo | 83baba5 | 2023-09-11 20:17:40 -0400 | [diff] [blame] | 21 | |
| 22 | pub use stream_config::*; |
| 23 | |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 24 | use nativewindow::*; |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 25 | use std::time::Instant; |
| 26 | |
Carlos Martinez Romero | 43d9afc | 2023-07-07 22:58:51 +0000 | [diff] [blame] | 27 | /// This function will print Hello World. |
| 28 | #[no_mangle] |
| 29 | pub extern "C" fn hello() -> bool { |
| 30 | println!("Hello world."); |
| 31 | true |
| 32 | } |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 33 | |
| 34 | /// BufferPublishers provide buffers to BufferSusbscribers. Depending on the |
| 35 | /// particular object in question, these could be allocated locally or provided |
| 36 | /// over IPC. |
| 37 | /// |
| 38 | /// BufferPublishers are required to adhere to the following, based on the |
| 39 | /// reactive streams specification: |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 40 | /// * The total number of on_next´s signalled by a Publisher to a Subscriber |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 41 | /// MUST be less than or equal to the total number of elements requested by that |
| 42 | /// Subscriber´s Subscription at all times. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 43 | /// * A Publisher MAY signal fewer on_next than requested and terminate the |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 44 | /// Subscription by calling on_complete or on_error. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 45 | /// * on_subscribe, on_next, on_error and on_complete signaled to a Subscriber |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 46 | /// MUST be signaled serially. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 47 | /// * If a Publisher fails it MUST signal an on_error. |
| 48 | /// * If a Publisher terminates successfully (finite stream) it MUST signal an |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 49 | /// on_complete. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 50 | /// * If a Publisher signals either on_error or on_complete on a Subscriber, |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 51 | /// that Subscriber’s Subscription MUST be considered cancelled. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 52 | /// * Once a terminal state has been signaled (on_error, on_complete) it is |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 53 | /// REQUIRED that no further signals occur. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 54 | /// * If a Subscription is cancelled its Subscriber MUST eventually stop being |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 55 | /// signaled. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 56 | /// * A Publisher MAY support multiple Subscribers and decides whether each |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 57 | /// Subscription is unicast or multicast. |
| 58 | pub trait BufferPublisher { |
James Shargo | 83baba5 | 2023-09-11 20:17:40 -0400 | [diff] [blame] | 59 | /// Returns the StreamConfig of buffers that publisher creates. |
| 60 | fn get_publisher_stream_config(&self) -> StreamConfig; |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 61 | /// This function will create the subscription between the publisher and |
| 62 | /// the subscriber. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 63 | fn subscribe(&mut self, subscriber: impl BufferSubscriber + 'static); |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 64 | } |
| 65 | |
| 66 | /// BufferSubscribers can subscribe to BufferPublishers. They can request Frames |
| 67 | /// via the BufferSubscription they get from the publisher, then receive Frames |
| 68 | /// via on_next. |
| 69 | /// |
| 70 | /// BufferSubcribers are required to adhere to the following, based on the |
| 71 | /// reactive streams specification: |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 72 | /// * The total number of on_next´s signalled by a Publisher to a Subscriber |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 73 | /// MUST be less than or equal to the total number of elements requested by that |
| 74 | /// Subscriber´s Subscription at all times. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 75 | /// * A Publisher MAY signal fewer on_next than requested and terminate the |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 76 | /// Subscription by calling on_complete or on_error. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 77 | /// * on_subscribe, on_next, on_error and on_complete signaled to a Subscriber |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 78 | /// MUST be signaled serially. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 79 | /// * If a Publisher fails it MUST signal an on_error. |
| 80 | /// * If a Publisher terminates successfully (finite stream) it MUST signal an |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 81 | /// on_complete. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 82 | /// * If a Publisher signals either on_error or on_complete on a Subscriber, |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 83 | /// that Subscriber’s Subscription MUST be considered cancelled. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 84 | /// * Once a terminal state has been signaled (on_error, on_complete) it is |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 85 | /// REQUIRED that no further signals occur. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 86 | /// * If a Subscription is cancelled its Subscriber MUST eventually stop being |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 87 | /// signaled. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 88 | /// * Publisher.subscribe MAY be called as many times as wanted but MUST be |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 89 | /// with a different Subscriber each time. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 90 | /// * A Publisher MAY support multiple Subscribers and decides whether each |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 91 | /// Subscription is unicast or multicast. |
| 92 | pub trait BufferSubscriber { |
James Shargo | 83baba5 | 2023-09-11 20:17:40 -0400 | [diff] [blame] | 93 | /// The StreamConfig of buffers that this subscriber expects. |
| 94 | fn get_subscriber_stream_config(&self) -> StreamConfig; |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 95 | /// This function will be called at the beginning of the subscription. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 96 | fn on_subscribe(&mut self, subscription: Box<dyn BufferSubscription>); |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 97 | /// This function will be called for buffer that comes in. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 98 | fn on_next(&mut self, frame: Frame); |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 99 | /// This function will be called in case of an error. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 100 | fn on_error(&mut self, error: BufferError); |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 101 | /// This function will be called on finite streams when done. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 102 | fn on_complete(&mut self); |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 103 | } |
| 104 | |
| 105 | /// BufferSubscriptions serve as the bridge between BufferPublishers and |
| 106 | /// BufferSubscribers. BufferSubscribers receive a BufferSubscription when they |
| 107 | /// subscribe to a BufferPublisher via on_subscribe. |
| 108 | /// This object is to be used by the BufferSubscriber to cancel its subscription |
| 109 | /// or request more buffers. |
| 110 | /// |
| 111 | /// BufferSubcriptions are required to adhere to the following, based on the |
| 112 | /// reactive streams specification: |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 113 | /// * Subscription.request and Subscription.cancel MUST only be called inside |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 114 | /// of its Subscriber context. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 115 | /// * The Subscription MUST allow the Subscriber to call Subscription.request |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 116 | /// synchronously from within on_next or on_subscribe. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 117 | /// * Subscription.request MUST place an upper bound on possible synchronous |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 118 | /// recursion between Publisher and Subscriber. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 119 | /// * Subscription.request SHOULD respect the responsivity of its caller by |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 120 | /// returning in a timely manner. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 121 | /// * Subscription.cancel MUST respect the responsivity of its caller by |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 122 | /// returning in a timely manner, MUST be idempotent and MUST be thread-safe. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 123 | /// * After the Subscription is cancelled, additional |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 124 | /// Subscription.request(n: u64) MUST be NOPs. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 125 | /// * After the Subscription is cancelled, additional Subscription.cancel() |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 126 | /// MUST be NOPs. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 127 | /// * While the Subscription is not cancelled, Subscription.request(n: u64) |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 128 | /// MUST register the given number of additional elements to be produced to the |
| 129 | /// respective subscriber. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 130 | /// * While the Subscription is not cancelled, Subscription.request(n: u64) |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 131 | /// MUST signal on_error if the argument is <= 0. The cause message SHOULD |
| 132 | /// explain that non-positive request signals are illegal. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 133 | /// * While the Subscription is not cancelled, Subscription.request(n: u64) |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 134 | /// MAY synchronously call on_next on this (or other) subscriber(s). |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 135 | /// * While the Subscription is not cancelled, Subscription.request(n: u64) |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 136 | /// MAY synchronously call on_complete or on_error on this (or other) |
| 137 | /// subscriber(s). |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 138 | /// * While the Subscription is not cancelled, Subscription.cancel() MUST |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 139 | /// request the Publisher to eventually stop signaling its Subscriber. The |
| 140 | /// operation is NOT REQUIRED to affect the Subscription immediately. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 141 | /// * While the Subscription is not cancelled, Subscription.cancel() MUST |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 142 | /// request the Publisher to eventually drop any references to the corresponding |
| 143 | /// subscriber. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 144 | /// * While the Subscription is not cancelled, calling Subscription.cancel MAY |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 145 | /// cause the Publisher, if stateful, to transition into the shut-down state if |
| 146 | /// no other Subscription exists at this point. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 147 | /// * Calling Subscription.cancel MUST return normally. |
| 148 | /// * Calling Subscription.request MUST return normally. |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 149 | pub trait BufferSubscription { |
| 150 | /// request |
| 151 | fn request(&self, n: u64); |
| 152 | /// cancel |
| 153 | fn cancel(&self); |
| 154 | } |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 155 | |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 156 | /// Type used to describe errors produced by subscriptions. |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 157 | pub type BufferError = anyhow::Error; |
Carlos Martinez Romero | a400469 | 2023-07-20 21:34:15 +0000 | [diff] [blame] | 158 | |
| 159 | /// Struct used to contain the buffer. |
| 160 | pub struct Frame { |
| 161 | /// A handle to the C buffer interface. |
| 162 | pub buffer: AHardwareBuffer, |
| 163 | /// The time at which the buffer was dispatched. |
| 164 | pub present_time: Instant, |
| 165 | /// A fence used for reading/writing safely. |
| 166 | pub fence: i32, |
| 167 | } |
James Shargo | 77e3c54 | 2023-09-12 11:58:07 -0400 | [diff] [blame^] | 168 | |
| 169 | #[cfg(test)] |
| 170 | mod test { |
| 171 | #![allow(warnings, unused)] |
| 172 | use super::*; |
| 173 | |
| 174 | use anyhow::anyhow; |
| 175 | use std::borrow::BorrowMut; |
| 176 | use std::error::Error; |
| 177 | use std::ops::Add; |
| 178 | use std::sync::Arc; |
| 179 | use std::time::Duration; |
| 180 | |
| 181 | use crate::publishers::testing::*; |
| 182 | use crate::subscribers::{testing::*, SharedSubscriber}; |
| 183 | |
| 184 | const STREAM_CONFIG: StreamConfig = StreamConfig { |
| 185 | width: 1, |
| 186 | height: 1, |
| 187 | layers: 1, |
| 188 | format: AHardwareBuffer_Format::AHARDWAREBUFFER_FORMAT_R8G8B8A8_UNORM, |
| 189 | usage: AHardwareBuffer_UsageFlags::AHARDWAREBUFFER_USAGE_CPU_READ_OFTEN, |
| 190 | stride: 0, |
| 191 | }; |
| 192 | |
| 193 | fn make_frame() -> Frame { |
| 194 | Frame { |
| 195 | buffer: STREAM_CONFIG |
| 196 | .create_hardware_buffer() |
| 197 | .expect("Unable to create hardware buffer for test"), |
| 198 | present_time: Instant::now() + Duration::from_secs(1), |
| 199 | fence: 0, |
| 200 | } |
| 201 | } |
| 202 | |
| 203 | #[test] |
| 204 | fn test_test_implementations_next() { |
| 205 | let subscriber = SharedSubscriber::new(TestSubscriber::new(STREAM_CONFIG)); |
| 206 | let mut publisher = TestPublisher::new(STREAM_CONFIG); |
| 207 | |
| 208 | publisher.subscribe(subscriber.clone()); |
| 209 | assert!(subscriber.map_inner(|s| s.has_subscription())); |
| 210 | assert!(publisher.has_subscriber()); |
| 211 | |
| 212 | publisher.send_frame(make_frame()); |
| 213 | let events = subscriber.map_inner_mut(|s| s.take_events()); |
| 214 | assert!(!matches!(events.last().unwrap(), TestingSubscriberEvent::Next(_))); |
| 215 | |
| 216 | subscriber.map_inner(|s| s.request(1)); |
| 217 | assert_eq!(publisher.pending_requests(), 1); |
| 218 | |
| 219 | publisher.send_frame(make_frame()); |
| 220 | let events = subscriber.map_inner_mut(|s| s.take_events()); |
| 221 | assert!(matches!(events.last().unwrap(), TestingSubscriberEvent::Next(_))); |
| 222 | assert_eq!(publisher.pending_requests(), 0); |
| 223 | } |
| 224 | |
| 225 | #[test] |
| 226 | fn test_test_implementations_complete() { |
| 227 | let subscriber = SharedSubscriber::new(TestSubscriber::new(STREAM_CONFIG)); |
| 228 | let mut publisher = TestPublisher::new(STREAM_CONFIG); |
| 229 | |
| 230 | publisher.subscribe(subscriber.clone()); |
| 231 | assert!(subscriber.map_inner(|s| s.has_subscription())); |
| 232 | assert!(publisher.has_subscriber()); |
| 233 | |
| 234 | publisher.send_complete(); |
| 235 | let events = subscriber.map_inner_mut(|s| s.take_events()); |
| 236 | assert!(matches!(events.last().unwrap(), TestingSubscriberEvent::Complete)); |
| 237 | } |
| 238 | |
| 239 | #[test] |
| 240 | fn test_test_implementations_error() { |
| 241 | let subscriber = SharedSubscriber::new(TestSubscriber::new(STREAM_CONFIG)); |
| 242 | let mut publisher = TestPublisher::new(STREAM_CONFIG); |
| 243 | |
| 244 | publisher.subscribe(subscriber.clone()); |
| 245 | assert!(subscriber.map_inner(|s| s.has_subscription())); |
| 246 | assert!(publisher.has_subscriber()); |
| 247 | |
| 248 | publisher.send_error(anyhow!("error")); |
| 249 | let events = subscriber.map_inner_mut(|s| s.take_events()); |
| 250 | assert!(matches!(events.last().unwrap(), TestingSubscriberEvent::Error(_))); |
| 251 | } |
| 252 | } |