Merge pull request lightningdevkit#91 from tnull/2023-05-add-try-next-event

Introduce non-blocking event queue access

Author: tnull

bindings/kotlin/ldk-node-jvm/lib/src/test/kotlin/org/lightningdevkit/ldknode/LibraryTest.kt

@@ -123,12 +123,12 @@ class LibraryTest {
 
         node1.connectOpenChannel(nodeId2, listenAddress2, 50000u, null, true)
 
-        val channelPendingEvent1 = node1.nextEvent()
+        val channelPendingEvent1 = node1.waitNextEvent()
         println("Got event: $channelPendingEvent1")
         assert(channelPendingEvent1 is Event.ChannelPending)
         node1.eventHandled()
 
-        val channelPendingEvent2 = node2.nextEvent()
+        val channelPendingEvent2 = node2.waitNextEvent()
         println("Got event: $channelPendingEvent2")
         assert(channelPendingEvent2 is Event.ChannelPending)
         node2.eventHandled()
@@ -153,12 +153,12 @@ class LibraryTest {
         assert(spendableBalance1AfterOpen < 50000u)
         assertEquals(100000u, spendableBalance2AfterOpen)
 
-        val channelReadyEvent1 = node1.nextEvent()
+        val channelReadyEvent1 = node1.waitNextEvent()
         println("Got event: $channelReadyEvent1")
         assert(channelReadyEvent1 is Event.ChannelReady)
         node1.eventHandled()
 
-        val channelReadyEvent2 = node2.nextEvent()
+        val channelReadyEvent2 = node2.waitNextEvent()
         println("Got event: $channelReadyEvent2")
         assert(channelReadyEvent2 is Event.ChannelReady)
         node2.eventHandled()
@@ -172,24 +172,24 @@ class LibraryTest {
 
         node1.sendPayment(invoice)
 
-        val paymentSuccessfulEvent = node1.nextEvent()
+        val paymentSuccessfulEvent = node1.waitNextEvent()
         println("Got event: $paymentSuccessfulEvent")
         assert(paymentSuccessfulEvent is Event.PaymentSuccessful)
         node1.eventHandled()
 
-        val paymentReceivedEvent = node2.nextEvent()
+        val paymentReceivedEvent = node2.waitNextEvent()
         println("Got event: $paymentReceivedEvent")
         assert(paymentReceivedEvent is Event.PaymentReceived)
         node2.eventHandled()
 
         node2.closeChannel(channelId, nodeId1)
 
-        val channelClosedEvent1 = node1.nextEvent()
+        val channelClosedEvent1 = node1.waitNextEvent()
         println("Got event: $channelClosedEvent1")
         assert(channelClosedEvent1 is Event.ChannelClosed)
         node1.eventHandled()
 
-        val channelClosedEvent2 = node2.nextEvent()
+        val channelClosedEvent2 = node2.waitNextEvent()
         println("Got event: $channelClosedEvent2")
         assert(channelClosedEvent2 is Event.ChannelClosed)
         node2.eventHandled()

bindings/ldk_node.udl

@@ -21,7 +21,8 @@ interface Node {
 	void start();
 	[Throws=NodeError]
 	void stop();
-	Event next_event();
+	Event? next_event();
+	Event wait_next_event();
 	void event_handled();
 	PublicKey node_id();
 	SocketAddr? listening_address();

src/event.rs

@@ -141,7 +141,12 @@ where
 		Ok(())
 	}
 
-	pub(crate) fn next_event(&self) -> Event {
+	pub(crate) fn next_event(&self) -> Option<Event> {
+		let locked_queue = self.queue.lock().unwrap();
+		locked_queue.front().map(|e| e.clone())
+	}
+
+	pub(crate) fn wait_next_event(&self) -> Event {
 		let locked_queue =
 			self.notifier.wait_while(self.queue.lock().unwrap(), |queue| queue.is_empty()).unwrap();
 		locked_queue.front().unwrap().clone()
@@ -700,6 +705,7 @@ mod tests {
 		let store = Arc::new(TestStore::new());
 		let logger = Arc::new(TestLogger::new());
 		let event_queue = EventQueue::new(Arc::clone(&store), Arc::clone(&logger));
+		assert_eq!(event_queue.next_event(), None);
 
 		let expected_event = Event::ChannelReady {
 			channel_id: ChannelId([23u8; 32]),
@@ -710,7 +716,8 @@ mod tests {
 
 		// Check we get the expected event and that it is returned until we mark it handled.
 		for _ in 0..5 {
-			assert_eq!(event_queue.next_event(), expected_event);
+			assert_eq!(event_queue.wait_next_event(), expected_event);
+			assert_eq!(event_queue.next_event(), Some(expected_event.clone()));
 			assert_eq!(false, store.get_and_clear_did_persist());
 		}
 
@@ -720,7 +727,7 @@ mod tests {
 			.unwrap();
 		let deser_event_queue =
 			EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
-		assert_eq!(deser_event_queue.next_event(), expected_event);
+		assert_eq!(deser_event_queue.wait_next_event(), expected_event);
 		assert!(!store.get_and_clear_did_persist());
 
 		// Check we persisted on `event_handled()`

src/lib.rs

@@ -931,13 +931,24 @@ impl Node {
 		Ok(())
 	}
 
-	/// Blocks until the next event is available.
+	/// Returns the next event in the event queue, if currently available.
+	///
+	/// Will return `Some(..)` if an event is available and `None` otherwise.
 	///
 	/// **Note:** this will always return the same event until handling is confirmed via [`Node::event_handled`].
-	pub fn next_event(&self) -> Event {
+	pub fn next_event(&self) -> Option<Event> {
 		self.event_queue.next_event()
 	}
 
+	/// Returns the next event in the event queue.
+	///
+	/// Will block the current thread until the next event is available.
+	///
+	/// **Note:** this will always return the same event until handling is confirmed via [`Node::event_handled`].
+	pub fn wait_next_event(&self) -> Event {
+		self.event_queue.wait_next_event()
+	}
+
 	/// Confirm the last retrieved event handled.
 	///
 	/// **Note:** This **MUST** be called after each event has been handled.

src/test/functional_tests.rs

@@ -33,6 +33,10 @@ fn channel_full_cycle() {
 	assert_eq!(node_a.onchain_balance().unwrap().get_spendable(), premine_amount_sat);
 	assert_eq!(node_b.onchain_balance().unwrap().get_spendable(), premine_amount_sat);
 
+	// Check we haven't got any events yet
+	assert_eq!(node_a.next_event(), None);
+	assert_eq!(node_b.next_event(), None);
+
 	println!("\nA -- connect_open_channel -> B");
 	let funding_amount_sat = 80_000;
 	let push_msat = (funding_amount_sat / 2) * 1000; // balance the channel
@@ -49,9 +53,10 @@ fn channel_full_cycle() {
 	assert_eq!(node_a.list_peers().first().unwrap().node_id, node_b.node_id());
 	expect_event!(node_a, ChannelPending);
 
-	let funding_txo = match node_b.next_event() {
+	let funding_txo = match node_b.wait_next_event() {
 		ref e @ Event::ChannelPending { funding_txo, .. } => {
 			println!("{} got event {:?}", std::stringify!(node_b), e);
+			assert_eq!(node_b.next_event().as_ref(), Some(e));
 			node_b.event_handled();
 			funding_txo
 		}
@@ -77,7 +82,7 @@ fn channel_full_cycle() {
 
 	expect_event!(node_a, ChannelReady);
 
-	let ev = node_b.next_event();
+	let ev = node_b.wait_next_event();
 	let channel_id = match ev {
 		ref e @ Event::ChannelReady { ref channel_id, .. } => {
 			println!("{} got event {:?}", std::stringify!(node_b), e);
@@ -138,7 +143,7 @@ fn channel_full_cycle() {
 	let overpaid_amount_msat = invoice_amount_2_msat + 100;
 	let payment_hash = node_a.send_payment_using_amount(&invoice, overpaid_amount_msat).unwrap();
 	expect_event!(node_a, PaymentSuccessful);
-	let received_amount = match node_b.next_event() {
+	let received_amount = match node_b.wait_next_event() {
 		ref e @ Event::PaymentReceived { amount_msat, .. } => {
 			println!("{} got event {:?}", std::stringify!(node_b), e);
 			node_b.event_handled();
@@ -164,7 +169,7 @@ fn channel_full_cycle() {
 		node_a.send_payment_using_amount(&variable_amount_invoice, determined_amount_msat).unwrap();
 
 	expect_event!(node_a, PaymentSuccessful);
-	let received_amount = match node_b.next_event() {
+	let received_amount = match node_b.wait_next_event() {
 		ref e @ Event::PaymentReceived { amount_msat, .. } => {
 			println!("{} got event {:?}", std::stringify!(node_b), e);
 			node_b.event_handled();
@@ -202,6 +207,10 @@ fn channel_full_cycle() {
 	let expected_final_amount_node_b_sat = premine_amount_sat + sum_of_all_payments_sat;
 	assert_eq!(node_b.onchain_balance().unwrap().get_spendable(), expected_final_amount_node_b_sat);
 
+	// Check we handled all events
+	assert_eq!(node_a.next_event(), None);
+	assert_eq!(node_b.next_event(), None);
+
 	node_a.stop().unwrap();
 	println!("\nA stopped");
 	node_b.stop().unwrap();

src/test/utils.rs

@@ -27,7 +27,7 @@ use std::time::Duration;
 
 macro_rules! expect_event {
 	($node: expr, $event_type: ident) => {{
-		match $node.next_event() {
+		match $node.wait_next_event() {
 			ref e @ Event::$event_type { .. } => {
 				println!("{} got event {:?}", std::stringify!($node), e);
 				$node.event_handled();