-adapt to latest exchange API change

1 files changed, 402 insertions, 0 deletions

diff --git a/src/backend/taler-merchant-httpd_exchanges.c b/src/backend/taler-merchant-httpd_exchanges.c
index 090cef95..5bce19a4 100644
--- a/src/backend/taler-merchant-httpd_exchanges.c
+++ b/src/backend/taler-merchant-httpd_exchanges.c
@@ -104,6 +104,86 @@ struct TMH_EXCHANGES_FindOperation
 
 
 /**
+ * Information we keep for a pending #MMH_EXCHANGES_keys4exchange() operation.
+ */
+struct TMH_EXCHANGES_Find2Operation
+{
+
+  /**
+   * Kept in a DLL.
+   */
+  struct TMH_EXCHANGES_Find2Operation *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct TMH_EXCHANGES_Find2Operation *prev;
+
+  /**
+   * Function to call with the result.
+   */
+  TMH_EXCHANGES_Find2Continuation fc;
+
+  /**
+   * Closure for @e fc.
+   */
+  void *fc_cls;
+
+  /**
+   * Exchange we wait for the /keys for.
+   */
+  struct TMH_Exchange *my_exchange;
+
+  /**
+   * Task scheduled to asynchronously return the result to
+   * the find continuation.
+   */
+  struct GNUNET_SCHEDULER_Task *at;
+
+};
+
+
+/**
+ * Information we keep for a pending #MMH_EXCHANGES_wire4exchange() operation.
+ */
+struct TMH_EXCHANGES_WireOperation
+{
+
+  /**
+   * Kept in a DLL.
+   */
+  struct TMH_EXCHANGES_WireOperation *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct TMH_EXCHANGES_WireOperation *prev;
+
+  /**
+   * Function to call with the result.
+   */
+  TMH_EXCHANGES_WireContinuation fc;
+
+  /**
+   * Closure for @e fc.
+   */
+  void *fc_cls;
+
+  /**
+   * Exchange we wait for the /wire for.
+   */
+  struct TMH_Exchange *my_exchange;
+
+  /**
+   * Task scheduled to asynchronously return the result to
+   * the find continuation.
+   */
+  struct GNUNET_SCHEDULER_Task *at;
+
+};
+
+
+/**
  * Information about wire transfer fees of an exchange, by wire method.
  */
 struct FeesByWireMethod
@@ -240,6 +320,26 @@ struct TMH_Exchange
   struct TMH_EXCHANGES_FindOperation *fo_tail;
 
   /**
+   * Head of FOs pending for this exchange.
+   */
+  struct TMH_EXCHANGES_Find2Operation *fo2_head;
+
+  /**
+   * Tail of FOs pending for this exchange.
+   */
+  struct TMH_EXCHANGES_Find2Operation *fo2_tail;
+
+  /**
+   * Head of /wire pending for this exchange.
+   */
+  struct TMH_EXCHANGES_WireOperation *w_head;
+
+  /**
+   * Tail of /wire pending for this exchange.
+   */
+  struct TMH_EXCHANGES_WireOperation *w_tail;
+
+  /**
    * Head of accounts of this exchange.
    */
   struct ExchangeAccount *acc_head;
@@ -891,8 +991,20 @@ static bool
 process_find_operations (struct TMH_Exchange *exchange)
 {
   struct TMH_EXCHANGES_FindOperation *fn;
+  struct TMH_EXCHANGES_Find2Operation *fn2;
+  struct TMH_EXCHANGES_WireOperation *wn;
   struct GNUNET_TIME_Timestamp now;
 
+  fn2 = NULL;
+  for (struct TMH_EXCHANGES_Find2Operation *fo = exchange->fo2_head;
+       NULL != fo;
+       fo = fn2)
+  {
+    fo->fc (fo->fc_cls,
+            TALER_EXCHANGE_get_keys (exchange->conn));
+    fn2 = fo->next;
+    TMH_EXCHANGES_keys4exchange_cancel (fo);
+  }
   if (! exchange->have_wire)
     return true;
   now = GNUNET_TIME_timestamp_get ();
@@ -957,6 +1069,16 @@ process_find_operations (struct TMH_Exchange *exchange)
     fn = fo->next;
     TMH_EXCHANGES_find_exchange_cancel (fo);
   }
+  wn = NULL;
+  for (struct TMH_EXCHANGES_WireOperation *w = exchange->w_head;
+       NULL != w;
+       w = wn)
+  {
+    w->fc (w->fc_cls,
+           exchange);
+    wn = w->next;
+    TMH_EXCHANGES_wire4exchange_cancel (w);
+  }
   return false;
 }
 
@@ -992,6 +1114,7 @@ handle_wire_data (void *cls,
   if (MHD_HTTP_OK != wr->hr.http_status)
   {
     struct TMH_EXCHANGES_FindOperation *fo;
+    struct TMH_EXCHANGES_WireOperation *w;
 
     exchange->have_wire = false;
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
@@ -1007,6 +1130,12 @@ handle_wire_data (void *cls,
               false);
       TMH_EXCHANGES_find_exchange_cancel (fo);
     }
+    while (NULL != (w = exchange->w_head))
+    {
+      w->fc (w->fc_cls,
+             NULL);
+      TMH_EXCHANGES_wire4exchange_cancel (w);
+    }
     return;
   }
   keys = TALER_EXCHANGE_get_keys (exchange->conn);
@@ -1024,6 +1153,7 @@ handle_wire_data (void *cls,
   {
     /* Report hard failure to all callbacks! */
     struct TMH_EXCHANGES_FindOperation *fo;
+    struct TMH_EXCHANGES_Find2Operation *fo2;
     struct TALER_EXCHANGE_HttpResponse hrx = {
       .ec = ecx,
       .http_status = 0,
@@ -1040,6 +1170,12 @@ handle_wire_data (void *cls,
               false);
       TMH_EXCHANGES_find_exchange_cancel (fo);
     }
+    while (NULL != (fo2 = exchange->fo2_head))
+    {
+      fo2->fc (fo2->fc_cls,
+               NULL);
+      TMH_EXCHANGES_keys4exchange_cancel (fo2);
+    }
     return;
   }
   exchange->have_wire = true;
@@ -1148,6 +1284,10 @@ free_exchange_entry (struct TMH_Exchange *exchange)
   }
   GNUNET_assert (NULL == exchange->fo_head);
   GNUNET_assert (NULL == exchange->fo_tail);
+  GNUNET_assert (NULL == exchange->fo2_head);
+  GNUNET_assert (NULL == exchange->fo2_tail);
+  GNUNET_assert (NULL == exchange->w_head);
+  GNUNET_assert (NULL == exchange->w_tail);
   GNUNET_free (exchange->url);
   GNUNET_free (exchange);
 }
@@ -1165,6 +1305,8 @@ fail_and_retry (struct TMH_Exchange *exchange,
                 const struct TALER_EXCHANGE_HttpResponse *hr)
 {
   struct TMH_EXCHANGES_FindOperation *fo;
+  struct TMH_EXCHANGES_Find2Operation *fo2;
+  struct TMH_EXCHANGES_WireOperation *w;
 
   exchange->have_keys = false;
   if (NULL != exchange->wire_request)
@@ -1185,6 +1327,18 @@ fail_and_retry (struct TMH_Exchange *exchange,
             false);
     TMH_EXCHANGES_find_exchange_cancel (fo);
   }
+  while (NULL != (fo2 = exchange->fo2_head))
+  {
+    fo2->fc (fo2->fc_cls,
+             NULL);
+    TMH_EXCHANGES_keys4exchange_cancel (fo2);
+  }
+  while (NULL != (w = exchange->w_head))
+  {
+    w->fc (w->fc_cls,
+           NULL);
+    TMH_EXCHANGES_wire4exchange_cancel (w);
+  }
   if ( (NULL == exchange->fo_head) &&
        (TALER_EC_GENERIC_CONFIGURATION_INVALID == hr->ec) )
   {
@@ -1345,6 +1499,38 @@ return_result (void *cls)
 
 
 /**
+ * Task to return find operation result asynchronously to caller.
+ *
+ * @param cls a `struct TMH_EXCHANGES_Find2Operation`
+ */
+static void
+return_result2 (void *cls)
+{
+  struct TMH_EXCHANGES_Find2Operation *fo = cls;
+  struct TMH_Exchange *exchange = fo->my_exchange;
+
+  fo->at = NULL;
+  process_find_operations (exchange);
+}
+
+
+/**
+ * Task to return find operation result asynchronously to caller.
+ *
+ * @param cls a `struct TMH_EXCHANGES_WireOperation`
+ */
+static void
+return_wire_result (void *cls)
+{
+  struct TMH_EXCHANGES_WireOperation *w = cls;
+  struct TMH_Exchange *exchange = w->my_exchange;
+
+  w->at = NULL;
+  process_find_operations (exchange);
+}
+
+
+/**
  * Lookup exchange by @a exchange_url.
  *
  * @param exchange_url base URL to match against
@@ -1481,6 +1667,188 @@ TMH_EXCHANGES_find_exchange (const char *chosen_exchange,
 }
 
 
+struct TMH_EXCHANGES_Find2Operation *
+TMH_EXCHANGES_keys4exchange (
+  const char *chosen_exchange,
+  TMH_EXCHANGES_Find2Continuation fc,
+  void *fc_cls)
+{
+  struct TMH_Exchange *exchange;
+  struct TMH_EXCHANGES_Find2Operation *fo;
+
+  if (NULL == merchant_curl_ctx)
+  {
+    GNUNET_break (0);
+    return NULL;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Trying to find chosen exchange `%s'\n",
+              chosen_exchange);
+  /* Check if the exchange is known */
+  exchange = lookup_exchange (chosen_exchange);
+  if (NULL == exchange)
+  {
+    /* This is a new exchange */
+    exchange = GNUNET_new (struct TMH_Exchange);
+    exchange->url = GNUNET_strdup (chosen_exchange);
+    GNUNET_CONTAINER_DLL_insert (exchange_head,
+                                 exchange_tail,
+                                 exchange);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "The exchange `%s' is new\n",
+                chosen_exchange);
+  }
+
+  fo = GNUNET_new (struct TMH_EXCHANGES_Find2Operation);
+  fo->fc = fc;
+  fo->fc_cls = fc_cls;
+  fo->my_exchange = exchange;
+  GNUNET_CONTAINER_DLL_insert (exchange->fo2_head,
+                               exchange->fo2_tail,
+                               fo);
+
+  if (GNUNET_TIME_absolute_is_past (exchange->first_retry))
+  {
+    /* increment exponential-backoff */
+    exchange->retry_delay = RETRY_BACKOFF (exchange->retry_delay);
+    /* do not allow forced check until both backoff and #FORCED_RELOAD_DELAY
+       are satisfied again */
+    exchange->first_retry
+      = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_max (
+                                            exchange->retry_delay,
+                                            FORCED_RELOAD_DELAY));
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "/keys retry forced, waiting until %s\n",
+                GNUNET_TIME_absolute2s (exchange->first_retry));
+    if (NULL == exchange->conn)
+    {
+      /* Not connected at all yet */
+      exchange->retry_task
+        = GNUNET_SCHEDULER_add_now (&retry_exchange,
+                                    exchange);
+    }
+    else
+    {
+      /* Use existing connection, but update /keys */
+      exchange->keys_expiration
+        = TALER_EXCHANGE_check_keys_current (exchange->conn,
+                                             TALER_EXCHANGE_CKF_FORCE_DOWNLOAD,
+                                             NULL,
+                                             NULL);
+    }
+    return fo;
+  }
+
+  if (exchange->have_keys)
+  {
+    /* We are not currently waiting for a reply, immediately
+       return result */
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "The exchange `%s' is ready\n",
+                exchange->url);
+    GNUNET_assert (NULL == fo->at);
+    fo->at = GNUNET_SCHEDULER_add_now (&return_result2,
+                                       fo);
+    return fo;
+  }
+
+  /* If new or resumed, (re)try fetching /keys */
+  if ( (NULL == exchange->conn) &&
+       (NULL == exchange->retry_task) &&
+       (! exchange->have_keys) )
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Do not have current /keys data for `%s'. Will request /keys now\n",
+                exchange->url);
+    exchange->retry_task = GNUNET_SCHEDULER_add_now (&retry_exchange,
+                                                     exchange);
+    return fo;
+  }
+  /* No activity to launch, we are already doing so */
+  return fo;
+}
+
+
+struct TMH_EXCHANGES_WireOperation *
+TMH_EXCHANGES_wire4exchange (
+  const char *chosen_exchange,
+  TMH_EXCHANGES_WireContinuation fc,
+  void *fc_cls)
+{
+  struct TMH_Exchange *exchange;
+  struct TMH_EXCHANGES_WireOperation *w;
+
+  if (NULL == merchant_curl_ctx)
+  {
+    GNUNET_break (0);
+    return NULL;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Trying to find chosen exchange `%s'\n",
+              chosen_exchange);
+  /* Check if the exchange is known */
+  exchange = lookup_exchange (chosen_exchange);
+  if (NULL == exchange)
+  {
+    /* This is a new exchange */
+    exchange = GNUNET_new (struct TMH_Exchange);
+    exchange->url = GNUNET_strdup (chosen_exchange);
+    GNUNET_CONTAINER_DLL_insert (exchange_head,
+                                 exchange_tail,
+                                 exchange);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "The exchange `%s' is new\n",
+                chosen_exchange);
+  }
+
+  w = GNUNET_new (struct TMH_EXCHANGES_WireOperation);
+  w->fc = fc;
+  w->fc_cls = fc_cls;
+  w->my_exchange = exchange;
+  GNUNET_CONTAINER_DLL_insert (exchange->w_head,
+                               exchange->w_tail,
+                               w);
+  if (exchange->have_wire)
+  {
+    /* We are not currently waiting for a reply, immediately
+       return result */
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "The exchange `%s' is ready\n",
+                exchange->url);
+    GNUNET_assert (NULL == w->at);
+    w->at = GNUNET_SCHEDULER_add_now (&return_wire_result,
+                                      w);
+    return w;
+  }
+
+  if (NULL == exchange->wire_request)
+  {
+    if (NULL == exchange->conn)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Requesting /keys from `%s' to connect\n",
+                  exchange->url);
+      exchange->conn = TALER_EXCHANGE_connect (merchant_curl_ctx,
+                                               exchange->url,
+                                               &keys_mgmt_cb,
+                                               exchange,
+                                               TALER_EXCHANGE_OPTION_END);
+    }
+    else
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "Requesting /wire from `%s'\n",
+                  exchange->url);
+      exchange->wire_request
+        = TALER_EXCHANGE_wire (exchange->conn,
+                               &handle_wire_data,
+                               exchange);
+    }
+  }
+  return w;
+}
+
+
 void
 TMH_EXCHANGES_find_exchange_cancel (struct TMH_EXCHANGES_FindOperation *fo)
 {
@@ -1498,6 +1866,40 @@ TMH_EXCHANGES_find_exchange_cancel (struct TMH_EXCHANGES_FindOperation *fo)
 }
 
 
+void
+TMH_EXCHANGES_keys4exchange_cancel (struct TMH_EXCHANGES_Find2Operation *fo)
+{
+  struct TMH_Exchange *exchange = fo->my_exchange;
+
+  if (NULL != fo->at)
+  {
+    GNUNET_SCHEDULER_cancel (fo->at);
+    fo->at = NULL;
+  }
+  GNUNET_CONTAINER_DLL_remove (exchange->fo2_head,
+                               exchange->fo2_tail,
+                               fo);
+  GNUNET_free (fo);
+}
+
+
+void
+TMH_EXCHANGES_wire4exchange_cancel (struct TMH_EXCHANGES_WireOperation *w)
+{
+  struct TMH_Exchange *exchange = w->my_exchange;
+
+  if (NULL != w->at)
+  {
+    GNUNET_SCHEDULER_cancel (w->at);
+    w->at = NULL;
+  }
+  GNUNET_CONTAINER_DLL_remove (exchange->w_head,
+                               exchange->w_tail,
+                               w);
+  GNUNET_free (w);
+}
+
+
 /**
  * Function called on each configuration section. Finds sections
  * about exchanges, parses the entries and tries to connect to