Add a connect callback.

clients/c/cosmopolite.c

@@ -170,7 +170,21 @@ static void cosmo_handle_message(cosmo *instance, json_t *event) {
   }
 }
 
+static void cosmo_handle_connect(cosmo *instance) {
+  if (instance->connect_state == CONNECTED) {
+    return;
+  }
+  instance->connect_state = CONNECTED;
+  if (instance->callbacks.connect) {
+    instance->callbacks.connect(instance->passthrough);
+  }
+}
+
 static void cosmo_handle_logout(cosmo *instance, json_t *event) {
+  if (instance->login_state == LOGGED_OUT) {
+    return;
+  }
+  instance->login_state = LOGGED_OUT;
   if (instance->callbacks.logout) {
     instance->callbacks.logout(instance->passthrough);
   }
@@ -237,6 +251,8 @@ static json_t *cosmo_send_rpc(cosmo *instance, json_t *commands, json_t *ack) {
     instance->profile = strdup(profile);
   }
 
+  cosmo_handle_connect(instance);
+
   size_t index;
   json_t *event;
   json_array_foreach(events, index, event) {
@@ -508,6 +524,9 @@ cosmo *cosmo_create(const char *base_url, const char *client_id, const cosmo_cal
   assert(instance->subscriptions);
   instance->next_delay_ms = 0;
 
+  instance->connect_state = INITIAL_CONNECT;
+  instance->login_state = LOGIN_UNKNOWN;
+
   assert(!pthread_create(&instance->thread, NULL, cosmo_thread_main, instance));
   return instance;
 }

clients/c/cosmopolite.h

@@ -10,6 +10,7 @@
 #define COSMO_UUID_SIZE 37
 
 typedef struct {
+  void (*connect)(void *);
   void (*logout)(void *);
   void (*message)(const json_t *, void *);
 } cosmo_callbacks;
@@ -30,6 +31,18 @@ typedef struct {
   uint64_t next_delay_ms;
   unsigned int seedp;
 
+  enum {
+    INITIAL_CONNECT,
+    CONNECTED,
+    DISCONNECTED,
+  } connect_state;
+
+  enum {
+    LOGIN_UNKNOWN,
+    LOGGED_OUT,
+    LOGGED_IN,
+  } login_state;
+
   pthread_t thread;
   CURL *curl;
 } cosmo;

clients/c/test.c

@@ -16,9 +16,18 @@ typedef struct {
 
   const json_t *last_message;
   bool logout_fired;
+  bool connect_fired;
 } test_state;
 
 
+void on_connect(void *passthrough) {
+  test_state *state = passthrough;
+  assert(!pthread_mutex_lock(&state->lock));
+  state->connect_fired = true;
+  assert(!pthread_cond_signal(&state->cond));
+  assert(!pthread_mutex_unlock(&state->lock));
+}
+
 void on_logout(void *passthrough) {
   test_state *state = passthrough;
   assert(!pthread_mutex_lock(&state->lock));
@@ -57,6 +66,16 @@ void wait_for_logout(test_state *state) {
   assert(!pthread_mutex_unlock(&state->lock));
 }
 
+void wait_for_connect(test_state *state) {
+  assert(!pthread_mutex_lock(&state->lock));
+  while (!state->connect_fired) {
+    assert(!pthread_cond_wait(&state->cond, &state->lock));
+  }
+
+  state->connect_fired = false;
+  assert(!pthread_mutex_unlock(&state->lock));
+}
+
 test_state *create_test_state() {
   test_state *ret = malloc(sizeof(test_state));
   assert(ret);
@@ -65,6 +84,7 @@ test_state *create_test_state() {
   assert(!pthread_cond_init(&ret->cond, NULL));
   ret->last_message = NULL;
   ret->logout_fired = false;
+  ret->connect_fired = false;
   return ret;
 }
 
@@ -79,6 +99,7 @@ cosmo *create_client(test_state *state) {
   cosmo_uuid(client_id);
 
   cosmo_callbacks callbacks = {
+    .connect = on_connect,
     .logout = on_logout,
     .message = on_message,
   };
@@ -135,6 +156,13 @@ bool test_message_round_trip(test_state *state) {
   return true;
 }
 
+bool test_connect_fires(test_state *state) {
+  cosmo *client = create_client(state);
+  wait_for_connect(state);
+  cosmo_shutdown(client);
+  return true;
+}
+
 bool test_logout_fires(test_state *state) {
   cosmo *client = create_client(state);
   wait_for_logout(state);
@@ -171,8 +199,9 @@ bool test_resubscribe(test_state *state) {
 
 int main(int argc, char *argv[]) {
   RUN_TEST(test_create_destroy);
-  RUN_TEST(test_message_round_trip);
+  RUN_TEST(test_connect_fires);
   RUN_TEST(test_logout_fires);
+  RUN_TEST(test_message_round_trip);
   RUN_TEST(test_resubscribe);
 
   return 0;