feat: chat ffi status callback (tari-project#5583)

Description --- This PR adds a callback manager and the ability to pass a callback into the chat ffi for use when a contact's status changes. Motivation and Context --- Requested by the mobile team. Pattern similar to wallet ffi callbacks, although purposefully not copied, it just seemed to be the natural way to go 😄 How Has This Been Tested? --- Added integration test. Although we get 16 callback calls. Not _totally_ unexpected but more than I was... expecting. What process can a PR reviewer use to test or verify this change? --- Read some code.   Breaking Changes --- - [x] None - [ ] Requires data directory on base node to be deleted - [ ] Requires hard fork - [ ] Other - Please specify   --------- Co-authored-by: SW van Heerden <[email protected]>
brianp · Jul 13, 2023 · f68b85f · f68b85f
1 parent 5008d4f
commit f68b85f
Show file tree

Hide file tree

Showing 9 changed files with 193 additions and 6 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/base_layer/chat_ffi/Cargo.toml b/base_layer/chat_ffi/Cargo.toml
@@ -13,6 +13,7 @@ tari_common = { path = "../../common" }
 tari_common_types = { path = "../common_types" }
 tari_comms = { path = "../../comms/core" }
 tari_contacts = { path = "../contacts" }
+tari_shutdown = { path = "../../infrastructure/shutdown" }
 
 libc = "0.2.65"
 log = "0.4.6"

diff --git a/base_layer/chat_ffi/chat.h b/base_layer/chat_ffi/chat.h
@@ -12,8 +12,12 @@ struct ChatMessages;
 
 struct ClientFFI;
 
+struct ContactsLivenessData;
+
 struct TariAddress;
 
+typedef void (*CallbackContactStatusChange)(struct ContactsLivenessData*);
+
 #ifdef __cplusplus
 extern "C" {
 #endif // __cplusplus
@@ -35,7 +39,8 @@ extern "C" {
  */
 struct ClientFFI *create_chat_client(ApplicationConfig *config,
                                      const char *identity_file_path,
-                                     int *error_out);
+                                     int *error_out,
+                                     CallbackContactStatusChange callback_contact_status_change);
 
 /**
  * Frees memory for a ClientFFI

diff --git a/base_layer/chat_ffi/src/callback_handler.rs b/base_layer/chat_ffi/src/callback_handler.rs
@@ -0,0 +1,92 @@
+// Copyright 2023, The Tari Project
+//
+// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+// following conditions are met:
+//
+// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+// disclaimer.
+//
+// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+// following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+// products derived from this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use std::ops::Deref;
+
+use log::{debug, info, trace};
+use tari_contacts::contacts_service::handle::{ContactsLivenessData, ContactsLivenessEvent, ContactsServiceHandle};
+use tari_shutdown::ShutdownSignal;
+
+const LOG_TARGET: &str = "chat_ffi::callback_handler";
+
+pub(crate) type CallbackContactStatusChange = unsafe extern "C" fn(*mut ContactsLivenessData);
+
+#[derive(Clone)]
+pub struct CallbackHandler {
+    contacts_service_handle: ContactsServiceHandle,
+    callback_contact_status_change: CallbackContactStatusChange,
+    shutdown: ShutdownSignal,
+}
+
+impl CallbackHandler {
+    pub fn new(
+        contacts_service_handle: ContactsServiceHandle,
+        shutdown: ShutdownSignal,
+        callback_contact_status_change: CallbackContactStatusChange,
+    ) -> Self {
+        Self {
+            contacts_service_handle,
+            shutdown,
+            callback_contact_status_change,
+        }
+    }
+
+    pub(crate) async fn start(&mut self) {
+        let mut liveness_events = self.contacts_service_handle.get_contacts_liveness_event_stream();
+
+        loop {
+            tokio::select! {
+                event = liveness_events.recv() => {
+                    match event {
+                        Ok(liveness_event) => {
+                            match liveness_event.deref() {
+                                ContactsLivenessEvent::StatusUpdated(data) => {
+                                    trace!(target: LOG_TARGET,
+                                        "FFI Callback monitor received Contact Status Updated event"
+                                    );
+                                    self.trigger_contact_status_change(data.deref().clone());
+                                }
+                                ContactsLivenessEvent::NetworkSilence => {},
+                            }
+                        },
+                        Err(_) => { debug!(target: LOG_TARGET, "FFI Callback monitor had an error with contacts liveness")}
+                    }
+                },
+                _ = self.shutdown.wait() => {
+                    info!(target: LOG_TARGET, "ChatFFI Callback Handler shutting down because the shutdown signal was received");
+                    break;
+                },
+            }
+        }
+    }
+
+    fn trigger_contact_status_change(&mut self, data: ContactsLivenessData) {
+        debug!(
+            target: LOG_TARGET,
+            "Calling Contacts Liveness Data Updated callback function for contact {}",
+            data.address(),
+        );
+        unsafe {
+            (self.callback_contact_status_change)(Box::into_raw(Box::new(data)));
+        }
+    }
+}
diff --git a/base_layer/chat_ffi/src/lib.rs b/base_layer/chat_ffi/src/lib.rs
@@ -24,6 +24,7 @@
 
 use std::{ffi::CStr, path::PathBuf, ptr, str::FromStr, sync::Arc};
 
+use callback_handler::CallbackContactStatusChange;
 use libc::{c_char, c_int};
 use tari_app_utilities::identity_management::load_from_json;
 use tari_chat_client::{
@@ -37,8 +38,12 @@ use tari_comms::{multiaddr::Multiaddr, NodeIdentity};
 use tari_contacts::contacts_service::types::Message;
 use tokio::runtime::Runtime;
 
-use crate::error::{InterfaceError, LibChatError};
+use crate::{
+    callback_handler::CallbackHandler,
+    error::{InterfaceError, LibChatError},
+};
 
+mod callback_handler;
 mod error;
 
 #[derive(Clone)]
@@ -67,6 +72,7 @@ pub unsafe extern "C" fn create_chat_client(
     config: *mut ApplicationConfig,
     identity_file_path: *const c_char,
     error_out: *mut c_int,
+    callback_contact_status_change: CallbackContactStatusChange,
 ) -> *mut ClientFFI {
     let mut error = 0;
     ptr::swap(error_out, &mut error as *mut c_int);
@@ -112,6 +118,16 @@ pub unsafe extern "C" fn create_chat_client(
     let mut client = Client::new(identity, (*config).clone());
     runtime.block_on(client.initialize());
 
+    let mut callback_handler = CallbackHandler::new(
+        client.contacts.clone().expect("No contacts service loaded yet"),
+        client.shutdown.to_signal(),
+        callback_contact_status_change,
+    );
+
+    runtime.spawn(async move {
+        callback_handler.start().await;
+    });
+
     let client_ffi = ClientFFI { client, runtime };
 
     Box::into_raw(Box::new(client_ffi))

diff --git a/integration_tests/log4rs/cucumber.yml b/integration_tests/log4rs/cucumber.yml
@@ -127,6 +127,10 @@ loggers:
     appenders:
       - base_layer_base_node
     additive: false
+  chat_ffi:
+    level: debug
+    appenders:
+      - base_layer_contacts
   wallet:
     level: debug
     appenders:

diff --git a/integration_tests/src/chat_ffi.rs b/integration_tests/src/chat_ffi.rs
@@ -25,7 +25,7 @@ use std::{
     ffi::{c_void, CString},
     path::PathBuf,
     str::FromStr,
-    sync::{Arc, Mutex},
+    sync::{Arc, Mutex, Once},
 };
 
 use async_trait::async_trait;
@@ -45,13 +45,19 @@ use tari_contacts::contacts_service::{service::ContactOnlineStatus, types::Messa
 
 use crate::{chat_client::test_config, get_port};
 
+extern "C" fn callback_contact_status_change(_state: *mut c_void) {
+    let callback = ChatCallback::instance();
+    *callback.contact_status_change.lock().unwrap() += 1;
+}
+
 #[cfg_attr(windows, link(name = "tari_chat_ffi.dll"))]
 #[cfg_attr(not(windows), link(name = "tari_chat_ffi"))]
 extern "C" {
     pub fn create_chat_client(
         config: *mut c_void,
         identity_file_path: *const c_char,
         out_error: *const c_int,
+        callback_contact_status_change: unsafe extern "C" fn(*mut c_void),
     ) -> *mut ClientFFI;
     pub fn send_message(client: *mut ClientFFI, receiver: *mut c_void, message: *const c_char, out_error: *const c_int);
     pub fn add_contact(client: *mut ClientFFI, address: *mut c_void, out_error: *const c_int);
@@ -180,11 +186,37 @@ pub async fn spawn_ffi_chat_client(name: &str, seed_peers: Vec<Peer>, base_dir:
     let out_error = Box::into_raw(Box::new(0));
 
     unsafe {
-        client_ptr = create_chat_client(config_ptr, identity_path_c_char, out_error);
+        *ChatCallback::instance().contact_status_change.lock().unwrap() = 0;
+
+        client_ptr = create_chat_client(
+            config_ptr,
+            identity_path_c_char,
+            out_error,
+            callback_contact_status_change,
+        );
     }
 
     ChatFFI {
         ptr: Arc::new(Mutex::new(PtrWrapper(client_ptr))),
         identity,
     }
 }
+
+static mut INSTANCE: Option<ChatCallback> = None;
+static START: Once = Once::new();
+
+#[derive(Default)]
+pub struct ChatCallback {
+    pub contact_status_change: Mutex<u64>,
+}
+
+impl ChatCallback {
+    pub fn instance() -> &'static mut Self {
+        unsafe {
+            START.call_once(|| {
+                INSTANCE = Some(Self::default());
+            });
+            INSTANCE.as_mut().unwrap()
+        }
+    }
+}
diff --git a/integration_tests/tests/features/ChatFFI.feature b/integration_tests/tests/features/ChatFFI.feature
@@ -11,6 +11,16 @@ Feature: Chat FFI messaging
     When I use CHAT_A to send a message 'Hey there' to CHAT_B
     Then CHAT_B will have 1 message with CHAT_A
 
+  # Also flaky on CI. Seems liveness has issues on CI
+  @broken
+  Scenario: Callback for status change is received
+    Given I have a seed node SEED_A
+    When I have a chat FFI client CHAT_A connected to seed node SEED_A
+    When I have a chat FFI client CHAT_B connected to seed node SEED_A
+    When CHAT_A adds CHAT_B as a contact
+    When CHAT_A waits for contact CHAT_B to be online
+    Then there will be a contact status update callback of at least 1
+
   #This is flaky, passes on local run time, but fails CI
   @broken
   Scenario: A message is sent directly between two FFI clients

diff --git a/integration_tests/tests/steps/chat_ffi_steps.rs b/integration_tests/tests/steps/chat_ffi_steps.rs
@@ -20,8 +20,15 @@
 //   WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //   USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use cucumber::when;
-use tari_integration_tests::{chat_ffi::spawn_ffi_chat_client, TariWorld};
+use std::time::Duration;
+
+use cucumber::{then, when};
+use tari_integration_tests::{
+    chat_ffi::{spawn_ffi_chat_client, ChatCallback},
+    TariWorld,
+};
+
+use crate::steps::{HALF_SECOND, TWO_MINUTES_WITH_HALF_SECOND_SLEEP};
 
 #[when(expr = "I have a chat FFI client {word} connected to seed node {word}")]
 async fn chat_ffi_client_connected_to_base_node(world: &mut TariWorld, name: String, seed_node_name: String) {
@@ -35,3 +42,22 @@ async fn chat_ffi_client_connected_to_base_node(world: &mut TariWorld, name: Str
     .await;
     world.chat_clients.insert(name, Box::new(client));
 }
+
+#[then(expr = "there will be a contact status update callback of at least {int}")]
+async fn contact_status_update_callback(_world: &mut TariWorld, callback_count: usize) {
+    let mut count = 0;
+    for _ in 0..(TWO_MINUTES_WITH_HALF_SECOND_SLEEP) {
+        count = *ChatCallback::instance().contact_status_change.lock().unwrap();
+
+        if count >= callback_count as u64 {
+            return;
+        }
+
+        tokio::time::sleep(Duration::from_millis(HALF_SECOND)).await;
+    }
+
+    panic!(
+        "contact status update never received. Callbacks expected: {}, Callbacks received: {:?}",
+        callback_count, count
+    );
+}