CPP-402 - Add callback API for host events

Author: Michael Fero

cpp-driver/examples/host_listener/.gitignore

@@ -0,0 +1 @@
+host_listener

cpp-driver/examples/host_listener/CMakeLists.txt

@@ -0,0 +1,15 @@
+cmake_minimum_required(VERSION 2.6.4)
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ".")
+set(PROJECT_EXAMPLE_NAME host_listener)
+
+file(GLOB EXAMPLE_SRC_FILES ${CASS_ROOT_DIR}/examples/host_listener/*.c)
+include_directories(${INCLUDES})
+add_executable(${PROJECT_EXAMPLE_NAME} ${EXAMPLE_SRC_FILES})
+target_link_libraries(${PROJECT_EXAMPLE_NAME} ${PROJECT_LIB_NAME_TARGET} ${CASS_LIBS})
+add_dependencies(${PROJECT_EXAMPLE_NAME} ${PROJECT_LIB_NAME_TARGET})
+
+set_property(
+  TARGET ${PROJECT_EXAMPLE_NAME}
+  APPEND PROPERTY COMPILE_FLAGS ${CASS_EXAMPLE_C_FLAGS})
+set_property(TARGET ${PROJECT_EXAMPLE_NAME} PROPERTY FOLDER "Examples")

cpp-driver/examples/host_listener/host_listener.c

@@ -0,0 +1,113 @@
+/*
+  This is free and unencumbered software released into the public domain.
+
+  Anyone is free to copy, modify, publish, use, compile, sell, or
+  distribute this software, either in source code form or as a compiled
+  binary, for any purpose, commercial or non-commercial, and by any
+  means.
+
+  In jurisdictions that recognize copyright laws, the author or authors
+  of this software dedicate any and all copyright interest in the
+  software to the public domain. We make this dedication for the benefit
+  of the public at large and to the detriment of our heirs and
+  successors. We intend this dedication to be an overt act of
+  relinquishment in perpetuity of all present and future rights to this
+  software under copyright law.
+
+  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+  EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+  MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+  IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+  OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+  ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+  OTHER DEALINGS IN THE SOFTWARE.
+
+  For more information, please refer to <http://unlicense.org/>
+*/
+
+#include <cassandra.h>
+
+
+#include <stdio.h>
+#include <uv.h>
+
+void on_signal(uv_signal_t* handle, int signum) {
+  uv_signal_stop(handle);
+}
+
+void on_host_listener(CassHostListenerEvent event, CassInet inet, void* data);
+
+void print_error(CassFuture* future) {
+  const char* message;
+  size_t message_length;
+  cass_future_error_message(future, &message, &message_length);
+  fprintf(stderr, "Error: %.*s\n", (int)message_length, message);
+}
+
+CassCluster* create_cluster(const char* hosts) {
+  CassCluster* cluster = cass_cluster_new();
+  cass_log_set_level(CASS_LOG_DISABLED);
+  cass_cluster_set_contact_points(cluster, hosts);
+  cass_cluster_set_host_listener_callback(cluster, on_host_listener, NULL);
+  return cluster;
+}
+
+CassError connect_session(CassSession* session, const CassCluster* cluster) {
+  CassError rc = CASS_OK;
+  CassFuture* future = cass_session_connect(session, cluster);
+
+  cass_future_wait(future);
+  rc = cass_future_error_code(future);
+  if (rc != CASS_OK) {
+    print_error(future);
+  }
+  cass_future_free(future);
+
+  return rc;
+}
+
+void on_host_listener(CassHostListenerEvent event, CassInet inet, void* data) {
+  char address[CASS_INET_STRING_LENGTH];
+
+  cass_inet_string(inet, address);
+  if (event == CASS_HOST_LISTENER_EVENT_ADD) {
+    printf("Host %s has been ADDED\n", address);
+   } else if (event == CASS_HOST_LISTENER_EVENT_REMOVE) {
+    printf("Host %s has been REMOVED\n", address);
+   } else if (event == CASS_HOST_LISTENER_EVENT_UP) {
+    printf("Host %s is UP\n", address);
+   } else if (event == CASS_HOST_LISTENER_EVENT_DOWN) {
+    printf("Host %s is DOWN\n", address);
+   }
+}
+
+int main(int argc, char* argv[]) {
+  CassCluster* cluster = NULL;
+  CassSession* session = cass_session_new();
+  char* hosts = "127.0.0.1";
+  uv_loop_t loop;
+  uv_signal_t signal;
+  if (argc > 1) {
+    hosts = argv[1];
+  }
+
+  cluster = create_cluster(hosts);
+
+  if (connect_session(session, cluster) != CASS_OK) {
+    cass_cluster_free(cluster);
+    cass_session_free(session);
+    return -1;
+  }
+
+  uv_loop_init(&loop);
+  uv_signal_init(&loop, &signal);
+  uv_signal_start(&signal, on_signal, SIGINT);
+  fprintf(stderr, "Press CTRL+C to exit ...\n");
+  uv_run(&loop, UV_RUN_DEFAULT);
+  uv_loop_close(&loop);
+
+  cass_cluster_free(cluster);
+  cass_session_free(session);
+
+  return 0;
+}

cpp-driver/include/cassandra.h

@@ -809,7 +809,6 @@ typedef void (*CassFreeFunction)(void* ptr);
  */
 typedef struct CassAuthenticator_ CassAuthenticator;
 
-
 /**
  * A callback used to initiate an authentication exchange.
  *
@@ -869,7 +868,6 @@ typedef void (*CassAuthenticatorSuccessCallback)(CassAuthenticator* auth,
 typedef void (*CassAuthenticatorCleanupCallback)(CassAuthenticator* auth,
                                                  void* data);
 
-
 /**
  * A callback used to cleanup resources.
  *
@@ -887,6 +885,25 @@ typedef struct CassAuthenticatorCallbacks_ {
   CassAuthenticatorCleanupCallback cleanup_callback;
 } CassAuthenticatorCallbacks;
 
+typedef enum CassHostListenerEvent_ {
+  CASS_HOST_LISTENER_EVENT_UP,
+  CASS_HOST_LISTENER_EVENT_DOWN,
+  CASS_HOST_LISTENER_EVENT_ADD,
+  CASS_HOST_LISTENER_EVENT_REMOVE
+} CassHostListenerEvent;
+
+/**
+ * A callback used to indicate the host state for a node in the cluster.
+ *
+ * @param[in] event
+ * @param[in] address
+ * @param[in] data
+ * @see cass_cluster_set_host_listener_callback()
+ */
+typedef void(*CassHostListenerCallback)(CassHostListenerEvent event,
+                                        const CassInet address,
+                                        void* data);
+
 /***********************************************************************************
  *
  * Execution Profile
@@ -2655,6 +2672,24 @@ CASS_EXPORT CassError
 cass_cluster_set_no_compact(CassCluster* cluster,
                             cass_bool_t enabled);
 
+/**
+ * Sets a callback for handling host state changes in the cluster.
+ *
+ * <b>Note:</b> The callback is invoked only when state changes in the cluster
+ * are applicable to the configured load balancing policy(s).
+ *
+ * @public @memberor CassCluster
+ *
+ * @param[in] cluster
+ * @param[in] callback
+ * @param[in] data
+ * @return CASS_OK if successful, otherwise and error occurred
+ */
+CASS_EXPORT CassError
+cass_cluster_set_host_listener_callback(CassCluster* cluster,
+                                        CassHostListenerCallback callback,
+                                        void* data);
+
 /***********************************************************************************
  *
  * Session

cpp-driver/src/cluster.hpp

@@ -210,7 +210,7 @@ class Cluster : public RefCounted<Cluster>
    * Constructor. Don't use directly.
    *
    * @param connection The current control connection.
-   * @param listener A listener to handle events.
+   * @param listener A listener to handle cluster events.
    * @param event_loop The event loop.
    * @param connected_host The currently connected host.
    * @param hosts Available hosts for the cluster (based on load balancing

cpp-driver/src/cluster_config.cpp

@@ -434,7 +434,7 @@ void cass_cluster_set_use_schema(CassCluster* cluster,
 }
 
 CassError cass_cluster_set_use_hostname_resolution(CassCluster* cluster,
-                                              cass_bool_t enabled) {
+                                                   cass_bool_t enabled) {
   cluster->config().set_use_hostname_resolution(enabled == cass_true);
   return CASS_OK;
 }
@@ -479,9 +479,9 @@ CassError cass_cluster_set_execution_profile(CassCluster* cluster,
 }
 
 CassError cass_cluster_set_execution_profile_n(CassCluster* cluster,
-                                             const char* name,
-                                             size_t name_length,
-                                             CassExecProfile* profile) {
+                                              const char* name,
+                                              size_t name_length,
+                                              CassExecProfile* profile) {
   if (name_length == 0 || !profile) {
     return CASS_ERROR_LIB_BAD_PARAMS;
   }
@@ -503,13 +503,13 @@ CassError cass_cluster_set_prepare_on_up_or_add_host(CassCluster* cluster,
 }
 
 CassError cass_cluster_set_local_address(CassCluster* cluster,
-                                                     const char* name) {
+                                         const char* name) {
   return cass_cluster_set_local_address_n(cluster, name, SAFE_STRLEN(name));
 }
 
 CassError cass_cluster_set_local_address_n(CassCluster* cluster,
-                                                     const char* name,
-                                                     size_t name_length) {
+                                           const char* name,
+                                           size_t name_length) {
   cass::Address address;  // default to AF_UNSPEC
   if (name_length == 0 ||
       name == NULL ||
@@ -527,6 +527,15 @@ CassError cass_cluster_set_no_compact(CassCluster* cluster,
   return CASS_OK;
 }
 
+CassError cass_cluster_set_host_listener_callback(CassCluster* cluster,
+                                                  CassHostListenerCallback callback,
+                                                  void* data) {
+  cluster->config().set_host_listener(
+      cass::DefaultHostListener::Ptr(
+        cass::Memory::allocate<cass::ExternalHostListener>(callback, data)));
+  return CASS_OK;
+}
+
 void cass_cluster_free(CassCluster* cluster) {
   cass::Memory::deallocate(cluster->from());
 }

cpp-driver/src/config.hpp

@@ -67,7 +67,8 @@ class Config {
       , max_reusable_write_objects_(CASS_DEFAULT_MAX_REUSABLE_WRITE_OBJECTS)
       , prepare_on_all_hosts_(CASS_DEFAULT_PREPARE_ON_ALL_HOSTS)
       , prepare_on_up_or_add_host_(CASS_DEFAULT_PREPARE_ON_UP_OR_ADD_HOST)
-      , no_compact_(CASS_DEFAULT_NO_COMPACT) {
+      , no_compact_(CASS_DEFAULT_NO_COMPACT)
+      , host_listener_(Memory::allocate<DefaultHostListener>()) {
     profiles_.set_empty_key(String());
 
     // Assign the defaults to the cluster profile
@@ -397,6 +398,16 @@ class Config {
     application_version_ = application_version;
   }
 
+  const DefaultHostListener::Ptr& host_listener() const { return host_listener_; }
+  
+  void set_host_listener(const DefaultHostListener::Ptr& listener) {
+    if (listener) {
+      host_listener_ = listener;
+    } else {
+      host_listener_.reset(Memory::allocate<DefaultHostListener>());
+    }
+  }
+
 private:
   void init_profiles();
 
@@ -440,6 +451,7 @@ class Config {
   bool no_compact_;
   String application_name_;
   String application_version_;
+  DefaultHostListener::Ptr host_listener_;
 };
 
 } // namespace cass

cpp-driver/src/host.cpp

@@ -133,4 +133,33 @@ void Host::set(const Row* row, bool use_tokens) {
   }
 }
 
+ExternalHostListener::ExternalHostListener(const CassHostListenerCallback callback,
+                                           void *data)
+  : callback_(callback)
+  , data_(data) { }
+
+void ExternalHostListener::on_up(const Host::Ptr& host) {
+  CassInet address;
+  address.address_length = host->address().to_inet(address.address);
+  callback_(CASS_HOST_LISTENER_EVENT_UP, address, data_);
+}
+
+void ExternalHostListener::on_down(const Host::Ptr& host) {
+  CassInet address;
+  address.address_length = host->address().to_inet(address.address);
+  callback_(CASS_HOST_LISTENER_EVENT_DOWN, address, data_);
+}
+
+void ExternalHostListener::on_add(const Host::Ptr& host) {
+  CassInet address;
+  address.address_length = host->address().to_inet(address.address);
+  callback_(CASS_HOST_LISTENER_EVENT_ADD, address, data_);
+}
+
+void ExternalHostListener::on_remove(const Host::Ptr& host) {
+  CassInet address;
+  address.address_length = host->address().to_inet(address.address);
+  callback_(CASS_HOST_LISTENER_EVENT_REMOVE, address, data_);
+}
+
 } // namespace cass

cpp-driver/src/host.hpp

@@ -263,6 +263,34 @@ class HostListener {
   virtual void on_remove(const Host::Ptr& host) = 0;
 };
 
+class DefaultHostListener
+    : public HostListener
+    , public RefCounted<DefaultHostListener> {
+public:
+  typedef SharedRefPtr<DefaultHostListener> Ptr;
+
+  virtual void on_up(const Host::Ptr& host) { }
+  virtual void on_down(const Host::Ptr& host) { }
+  virtual void on_add(const Host::Ptr& host) { }
+  virtual void on_remove(const Host::Ptr& host) { }
+};
+
+class ExternalHostListener : public DefaultHostListener  {
+public:
+  typedef SharedRefPtr<ExternalHostListener> Ptr;
+  ExternalHostListener(const CassHostListenerCallback callback,
+                       void *data);
+
+  virtual void on_up(const Host::Ptr& host);
+  virtual void on_down(const Host::Ptr& host);
+  virtual void on_add(const Host::Ptr& host);
+  virtual void on_remove(const Host::Ptr& host);
+
+private:
+  const CassHostListenerCallback callback_;
+  void* data_;
+};
+
 typedef Map<Address, Host::Ptr> HostMap;
 
 struct GetAddress {