common.cpp

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <arrow-glib/arrow-glib.hpp>

#include <arrow-flight-glib/common.hpp>

G_BEGIN_DECLS

/**
 * SECTION: common
 * @section_id: common
 * @title: Classes both for client and server
 * @include: arrow-flight-glib/arrow-flight-glib.h
 *
 * #GAFlightCriteria is a class for criteria.
 *
 * #GAFlightLocation is a class for location.
 *
 * #GAFlightDescriptor is a base class for all descriptor classes such
 * as #GAFlightPathDescriptor.
 *
 * #GAFlightPathDescriptor is a class for path descriptor.
 *
 * #GAFlightCommandDescriptor is a class for command descriptor.
 *
 * #GAFlightTicket is a class for ticket.
 *
 * #GAFlightEndpoint is a class for endpoint.
 *
 * #GAFlightInfo is a class for flight information.
 *
 * #GAFlightStreamChunk is a class for a chunk in stream.
 *
 * #GAFlightRecordBatchReader is a class for reading record batches.
 *
 * Since: 5.0.0
 */

/**
 * GAFlightHeaderFunc:
 * @name: A header name.
 * @value: The value corresponding to the name.
 * @user_data: User data passed to gaflight_call_options_foreach_header()
 *   and so on.
 *
 * It is called with each header name/value pair, together with the
 * @user_data parameter which is passed to
 * gaflight_call_options_foreach_header() and so on.
 *
 * Since: 9.0.0
 */

typedef struct GAFlightCriteriaPrivate_
{
  arrow::flight::Criteria criteria;
  GBytes *expression;
} GAFlightCriteriaPrivate;

enum {
  PROP_EXPRESSION = 1,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightCriteria, gaflight_criteria, G_TYPE_OBJECT)

#define GAFLIGHT_CRITERIA_GET_PRIVATE(obj)                                               \
  static_cast<GAFlightCriteriaPrivate *>(                                                \
    gaflight_criteria_get_instance_private(GAFLIGHT_CRITERIA(obj)))

static void
gaflight_criteria_dispose(GObject *object)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(object);

  if (priv->expression) {
    g_bytes_unref(priv->expression);
    priv->expression = NULL;
  }

  G_OBJECT_CLASS(gaflight_criteria_parent_class)->dispose(object);
}

static void
gaflight_criteria_finalize(GObject *object)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(object);

  priv->criteria.~Criteria();

  G_OBJECT_CLASS(gaflight_criteria_parent_class)->finalize(object);
}

static void
gaflight_criteria_set_property(GObject *object,
                               guint prop_id,
                               const GValue *value,
                               GParamSpec *pspec)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_EXPRESSION:
    if (priv->expression) {
      g_bytes_unref(priv->expression);
    }
    priv->expression = static_cast<GBytes *>(g_value_dup_boxed(value));
    {
      gsize size;
      auto data = g_bytes_get_data(priv->expression, &size);
      priv->criteria.expression.assign(static_cast<const char *>(data), size);
    }
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_criteria_get_property(GObject *object,
                               guint prop_id,
                               GValue *value,
                               GParamSpec *pspec)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_EXPRESSION:
    g_value_set_boxed(value, priv->expression);
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_criteria_init(GAFlightCriteria *object)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(object);
  new (&priv->criteria) arrow::flight::Criteria;
}

static void
gaflight_criteria_class_init(GAFlightCriteriaClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->dispose = gaflight_criteria_dispose;
  gobject_class->finalize = gaflight_criteria_finalize;
  gobject_class->set_property = gaflight_criteria_set_property;
  gobject_class->get_property = gaflight_criteria_get_property;

  GParamSpec *spec;
  /**
   * GAFlightCriteria:expression:
   *
   * Opaque criteria expression, dependent on server implementation.
   *
   * Since: 5.0.0
   */
  spec = g_param_spec_boxed("expression",
                            "Expression",
                            "Opaque criteria expression, "
                            "dependent on server implementation",
                            G_TYPE_BYTES,
                            static_cast<GParamFlags>(G_PARAM_READWRITE));
  g_object_class_install_property(gobject_class, PROP_EXPRESSION, spec);
}

/**
 * gaflight_criteria_new:
 * @expression: A #GBytes.
 *
 * Returns: The newly created #GAFlightCriteria, %NULL on error.
 *
 * Since: 5.0.0
 */
GAFlightCriteria *
gaflight_criteria_new(GBytes *expression)
{
  return GAFLIGHT_CRITERIA(
    g_object_new(GAFLIGHT_TYPE_CRITERIA, "expression", expression, NULL));
}

typedef struct GAFlightLocationPrivate_
{
  arrow::flight::Location location;
} GAFlightLocationPrivate;

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightLocation, gaflight_location, G_TYPE_OBJECT)

#define GAFLIGHT_LOCATION_GET_PRIVATE(obj)                                               \
  static_cast<GAFlightLocationPrivate *>(                                                \
    gaflight_location_get_instance_private(GAFLIGHT_LOCATION(obj)))

static void
gaflight_location_finalize(GObject *object)
{
  auto priv = GAFLIGHT_LOCATION_GET_PRIVATE(object);

  priv->location.~Location();

  G_OBJECT_CLASS(gaflight_location_parent_class)->finalize(object);
}

static void
gaflight_location_init(GAFlightLocation *object)
{
  auto priv = GAFLIGHT_LOCATION_GET_PRIVATE(object);
  new (&priv->location) arrow::flight::Location;
}

static void
gaflight_location_class_init(GAFlightLocationClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->finalize = gaflight_location_finalize;
}

/**
 * gaflight_location_new:
 * @uri: An URI to specify location.
 * @error: (nullable): Return location for a #GError or %NULL.
 *
 * Returns: (nullable): The newly created location, %NULL on error.
 *
 * Since: 5.0.0
 */
GAFlightLocation *
gaflight_location_new(const gchar *uri, GError **error)
{
  auto location = GAFLIGHT_LOCATION(g_object_new(GAFLIGHT_TYPE_LOCATION, NULL));
  auto flight_location = gaflight_location_get_raw(location);
  if (garrow::check(error,
                    arrow::flight::Location::Parse(uri).Value(flight_location),
                    "[flight-location][new]")) {
    return location;
  } else {
    g_object_unref(location);
    return NULL;
  }
}

/**
 * gaflight_location_to_string:
 * @location: A #GAFlightLocation.
 *
 * Returns: A representation of this URI as a string.
 *
 *   It should be freed with g_free() when no longer needed.
 *
 * Since: 5.0.0
 */
gchar *
gaflight_location_to_string(GAFlightLocation *location)
{
  const auto flight_location = gaflight_location_get_raw(location);
  const auto string = flight_location->ToString();
  return g_strdup(string.c_str());
}

/**
 * gaflight_location_get_scheme:
 * @location: A #GAFlightLocation.
 *
 * Returns: The scheme of this URI.
 *
 *   It should be freed with g_free() when no longer needed.
 *
 * Since: 5.0.0
 */
gchar *
gaflight_location_get_scheme(GAFlightLocation *location)
{
  const auto flight_location = gaflight_location_get_raw(location);
  const auto scheme = flight_location->scheme();
  return g_strdup(scheme.c_str());
}

/**
 * gaflight_location_equal:
 * @location: A #GAFlightLocation.
 * @other_location: A #GAFlightLocation to be compared.
 *
 * Returns: %TRUE if both of them represents the same URI, %FALSE otherwise.
 *
 * Since: 5.0.0
 */
gboolean
gaflight_location_equal(GAFlightLocation *location, GAFlightLocation *other_location)
{
  const auto flight_location = gaflight_location_get_raw(location);
  const auto flight_other_location = gaflight_location_get_raw(other_location);
  return flight_location->Equals(*flight_other_location);
}

typedef struct GAFlightDescriptorPrivate_
{
  arrow::flight::FlightDescriptor descriptor;
} GAFlightDescriptorPrivate;

enum {
  PROP_DESCRIPTOR = 1,
};

G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE(GAFlightDescriptor,
                                    gaflight_descriptor,
                                    G_TYPE_OBJECT)

#define GAFLIGHT_DESCRIPTOR_GET_PRIVATE(obj)                                             \
  static_cast<GAFlightDescriptorPrivate *>(                                              \
    gaflight_descriptor_get_instance_private(GAFLIGHT_DESCRIPTOR(obj)))

static void
gaflight_descriptor_finalize(GObject *object)
{
  auto priv = GAFLIGHT_DESCRIPTOR_GET_PRIVATE(object);

  priv->descriptor.~FlightDescriptor();

  G_OBJECT_CLASS(gaflight_descriptor_parent_class)->finalize(object);
}

static void
gaflight_descriptor_set_property(GObject *object,
                                 guint prop_id,
                                 const GValue *value,
                                 GParamSpec *pspec)
{
  auto priv = GAFLIGHT_DESCRIPTOR_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_DESCRIPTOR:
    priv->descriptor =
      *static_cast<arrow::flight::FlightDescriptor *>(g_value_get_pointer(value));
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_descriptor_init(GAFlightDescriptor *object)
{
  auto priv = GAFLIGHT_DESCRIPTOR_GET_PRIVATE(object);
  new (&priv->descriptor) arrow::flight::FlightDescriptor;
}

static void
gaflight_descriptor_class_init(GAFlightDescriptorClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->finalize = gaflight_descriptor_finalize;
  gobject_class->set_property = gaflight_descriptor_set_property;

  GParamSpec *spec;
  spec = g_param_spec_pointer(
    "descriptor",
    "Descriptor",
    "The raw arrow::flight::FlightDescriptor",
    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
  g_object_class_install_property(gobject_class, PROP_EXPRESSION, spec);
}

/**
 * gaflight_descriptor_to_string:
 * @descriptor: A #GAFlightDescriptor.
 *
 * Returns: A descriptor as a string.
 *
 *   It should be freed with g_free() when no longer needed.
 *
 * Since: 5.0.0
 */
gchar *
gaflight_descriptor_to_string(GAFlightDescriptor *descriptor)
{
  const auto flight_descriptor = gaflight_descriptor_get_raw(descriptor);
  const auto string = flight_descriptor->ToString();
  return g_strdup(string.c_str());
}

/**
 * gaflight_descriptor_equal:
 * @descriptor: A #GAFlightDescriptor.
 * @other_descriptor: A #GAFlightDescriptor to be compared.
 *
 * Returns: %TRUE if both of them represents the same descriptor,
 *   %FALSE otherwise.
 *
 * Since: 5.0.0
 */
gboolean
gaflight_descriptor_equal(GAFlightDescriptor *descriptor,
                          GAFlightDescriptor *other_descriptor)
{
  const auto flight_descriptor = gaflight_descriptor_get_raw(descriptor);
  const auto flight_other_descriptor = gaflight_descriptor_get_raw(other_descriptor);
  return flight_descriptor->Equals(*flight_other_descriptor);
}

G_DEFINE_TYPE(GAFlightPathDescriptor, gaflight_path_descriptor, GAFLIGHT_TYPE_DESCRIPTOR)

static void
gaflight_path_descriptor_init(GAFlightPathDescriptor *object)
{
}

static void
gaflight_path_descriptor_class_init(GAFlightPathDescriptorClass *klass)
{
}

/**
 * gaflight_path_descriptor_new:
 * @paths: (array length=n_paths): List of paths identifying a
 *   particular dataset.
 * @n_paths: The number of @paths.
 *
 * Returns: The newly created #GAFlightPathDescriptor.
 *
 * Since: 5.0.0
 */
GAFlightPathDescriptor *
gaflight_path_descriptor_new(const gchar **paths, gsize n_paths)
{
  std::vector<std::string> flight_paths;
  for (gsize i = 0; i < n_paths; i++) {
    flight_paths.push_back(paths[i]);
  }
  auto flight_descriptor = arrow::flight::FlightDescriptor::Path(flight_paths);
  return GAFLIGHT_PATH_DESCRIPTOR(gaflight_descriptor_new_raw(&flight_descriptor));
}

/**
 * gaflight_path_descriptor_get_paths:
 * @descriptor: A #GAFlightPathDescriptor.
 *
 * Returns: (nullable) (array zero-terminated=1) (transfer full):
 *   The paths in this descriptor.
 *
 *   It must be freed with g_strfreev() when no longer needed.
 *
 * Since: 5.0.0
 */
gchar **
gaflight_path_descriptor_get_paths(GAFlightPathDescriptor *descriptor)
{
  const auto flight_descriptor =
    gaflight_descriptor_get_raw(GAFLIGHT_DESCRIPTOR(descriptor));
  const auto &flight_paths = flight_descriptor->path;
  if (flight_paths.empty()) {
    return NULL;
  } else {
    auto paths = g_new(gchar *, flight_paths.size() + 1);
    gsize i = 0;
    for (const auto &flight_path : flight_paths) {
      paths[i++] = g_strdup(flight_path.c_str());
    }
    paths[i] = NULL;
    return paths;
  }
}

G_DEFINE_TYPE(GAFlightCommandDescriptor,
              gaflight_command_descriptor,
              GAFLIGHT_TYPE_DESCRIPTOR)

static void
gaflight_command_descriptor_init(GAFlightCommandDescriptor *object)
{
}

static void
gaflight_command_descriptor_class_init(GAFlightCommandDescriptorClass *klass)
{
}

/**
 * gaflight_command_descriptor_new:
 * @command: Opaque value used to express a command.
 *
 * Returns: The newly created #GAFlightCommandDescriptor.
 *
 * Since: 5.0.0
 */
GAFlightCommandDescriptor *
gaflight_command_descriptor_new(const gchar *command)
{
  auto flight_descriptor = arrow::flight::FlightDescriptor::Command(command);
  return GAFLIGHT_COMMAND_DESCRIPTOR(gaflight_descriptor_new_raw(&flight_descriptor));
}

/**
 * gaflight_command_descriptor_get_command:
 * @descriptor: A #GAFlightCommandDescriptor.
 *
 * Returns: The opaque value used to express a command.
 *
 *   It should be freed with g_free() when no longer needed.
 *
 * Since: 5.0.0
 */
gchar *
gaflight_command_descriptor_get_command(GAFlightCommandDescriptor *descriptor)
{
  const auto flight_descriptor =
    gaflight_descriptor_get_raw(GAFLIGHT_DESCRIPTOR(descriptor));
  const auto &flight_command = flight_descriptor->cmd;
  return g_strdup(flight_command.c_str());
}

typedef struct GAFlightTicketPrivate_
{
  arrow::flight::Ticket ticket;
  GBytes *data;
} GAFlightTicketPrivate;

enum {
  PROP_DATA = 1,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightTicket, gaflight_ticket, G_TYPE_OBJECT)

#define GAFLIGHT_TICKET_GET_PRIVATE(obj)                                                 \
  static_cast<GAFlightTicketPrivate *>(                                                  \
    gaflight_ticket_get_instance_private(GAFLIGHT_TICKET(obj)))

static void
gaflight_ticket_dispose(GObject *object)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(object);

  if (priv->data) {
    g_bytes_unref(priv->data);
    priv->data = NULL;
  }

  G_OBJECT_CLASS(gaflight_ticket_parent_class)->dispose(object);
}

static void
gaflight_ticket_finalize(GObject *object)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(object);

  priv->ticket.~Ticket();

  G_OBJECT_CLASS(gaflight_ticket_parent_class)->finalize(object);
}

static void
gaflight_ticket_set_property(GObject *object,
                             guint prop_id,
                             const GValue *value,
                             GParamSpec *pspec)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_DATA:
    if (priv->data) {
      g_bytes_unref(priv->data);
    }
    priv->data = static_cast<GBytes *>(g_value_dup_boxed(value));
    {
      gsize size;
      auto data = g_bytes_get_data(priv->data, &size);
      priv->ticket.ticket.assign(static_cast<const char *>(data), size);
    }
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_ticket_get_property(GObject *object,
                             guint prop_id,
                             GValue *value,
                             GParamSpec *pspec)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_DATA:
    g_value_set_boxed(value, priv->data);
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_ticket_init(GAFlightTicket *object)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(object);
  new (&priv->ticket) arrow::flight::Ticket;
}

static void
gaflight_ticket_class_init(GAFlightTicketClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->dispose = gaflight_ticket_dispose;
  gobject_class->finalize = gaflight_ticket_finalize;
  gobject_class->set_property = gaflight_ticket_set_property;
  gobject_class->get_property = gaflight_ticket_get_property;

  GParamSpec *spec;
  /**
   * GAFlightTicket:data:
   *
   * Opaque identifier or credential to use when requesting a data
   * stream with the DoGet RPC.
   *
   * Since: 5.0.0
   */
  spec = g_param_spec_boxed("data",
                            "Data",
                            "Opaque identifier or credential to use "
                            "when requesting a data stream with the DoGet RPC",
                            G_TYPE_BYTES,
                            static_cast<GParamFlags>(G_PARAM_READWRITE));
  g_object_class_install_property(gobject_class, PROP_DATA, spec);
}

/**
 * gaflight_ticket_new:
 * @data: A #GBytes.
 *
 * Returns: The newly created #GAFlightTicket, %NULL on error.
 *
 * Since: 5.0.0
 */
GAFlightTicket *
gaflight_ticket_new(GBytes *data)
{
  return GAFLIGHT_TICKET(g_object_new(GAFLIGHT_TYPE_TICKET, "data", data, NULL));
}

/**
 * gaflight_ticket_equal:
 * @ticket: A #GAFlightTicket.
 * @other_ticket: A #GAFlightTicket to be compared.
 *
 * Returns: %TRUE if both of them represents the same ticket, %FALSE otherwise.
 *
 * Since: 5.0.0
 */
gboolean
gaflight_ticket_equal(GAFlightTicket *ticket, GAFlightTicket *other_ticket)
{
  const auto flight_ticket = gaflight_ticket_get_raw(ticket);
  const auto flight_other_ticket = gaflight_ticket_get_raw(other_ticket);
  return flight_ticket->Equals(*flight_other_ticket);
}

typedef struct GAFlightEndpointPrivate_
{
  arrow::flight::FlightEndpoint endpoint;
  GAFlightTicket *ticket;
  GList *locations;
} GAFlightEndpointPrivate;

enum {
  PROP_TICKET = 1,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightEndpoint, gaflight_endpoint, G_TYPE_OBJECT)

#define GAFLIGHT_ENDPOINT_GET_PRIVATE(obj)                                               \
  static_cast<GAFlightEndpointPrivate *>(                                                \
    gaflight_endpoint_get_instance_private(GAFLIGHT_ENDPOINT(obj)))

static void
gaflight_endpoint_dispose(GObject *object)
{
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(object);

  if (priv->ticket) {
    g_object_unref(priv->ticket);
    priv->ticket = NULL;
  }

  if (priv->locations) {
    g_list_free_full(priv->locations, g_object_unref);
    priv->locations = NULL;
  }

  G_OBJECT_CLASS(gaflight_endpoint_parent_class)->dispose(object);
}

static void
gaflight_endpoint_finalize(GObject *object)
{
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(object);

  priv->endpoint.~FlightEndpoint();

  G_OBJECT_CLASS(gaflight_endpoint_parent_class)->finalize(object);
}

static void
gaflight_endpoint_get_property(GObject *object,
                               guint prop_id,
                               GValue *value,
                               GParamSpec *pspec)
{
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_TICKET:
    g_value_set_object(value, priv->ticket);
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_endpoint_init(GAFlightEndpoint *object)
{
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(object);
  new (&priv->endpoint) arrow::flight::FlightEndpoint;
}

static void
gaflight_endpoint_class_init(GAFlightEndpointClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->dispose = gaflight_endpoint_dispose;
  gobject_class->finalize = gaflight_endpoint_finalize;
  gobject_class->get_property = gaflight_endpoint_get_property;

  GParamSpec *spec;
  /**
   * GAFlightEndpoint:ticket:
   *
   * Opaque ticket identify; use with DoGet RPC.
   *
   * Since: 5.0.0
   */
  spec = g_param_spec_object("ticket",
                             "Ticket",
                             "Opaque ticket identify; use with DoGet RPC",
                             GAFLIGHT_TYPE_TICKET,
                             static_cast<GParamFlags>(G_PARAM_READABLE));
  g_object_class_install_property(gobject_class, PROP_TICKET, spec);
}

/**
 * gaflight_endpoint_new:
 * @ticket: A #GAFlightTicket.
 * @locations: (element-type GAFlightLocation): A list of #GAFlightLocation.
 *
 * Returns: The newly created #GAFlightEndpoint, %NULL on error.
 *
 * Since: 5.0.0
 */
GAFlightEndpoint *
gaflight_endpoint_new(GAFlightTicket *ticket, GList *locations)
{
  auto endpoint = gaflight_endpoint_new_raw(nullptr, ticket);
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(endpoint);
  for (auto node = locations; node; node = node->next) {
    auto location = GAFLIGHT_LOCATION(node->data);
    priv->endpoint.locations.push_back(*gaflight_location_get_raw(location));
  }
  return endpoint;
}

/**
 * gaflight_endpoint_equal:
 * @endpoint: A #GAFlightEndpoint.
 * @other_endpoint: A #GAFlightEndpoint to be compared.
 *
 * Returns: %TRUE if both of them represents the same endpoint,
 *   %FALSE otherwise.
 *
 * Since: 5.0.0
 */
gboolean
gaflight_endpoint_equal(GAFlightEndpoint *endpoint, GAFlightEndpoint *other_endpoint)
{
  const auto flight_endpoint = gaflight_endpoint_get_raw(endpoint);
  const auto flight_other_endpoint = gaflight_endpoint_get_raw(other_endpoint);
  return flight_endpoint->Equals(*flight_other_endpoint);
}

/**
 * gaflight_endpoint_get_locations:
 * @endpoint: A #GAFlightEndpoint.
 *
 * Returns: (nullable) (element-type GAFlightLocation) (transfer full):
 *   The locations in this endpoint.
 *
 *   It must be freed with g_list_free() and g_object_unref() when no
 *   longer needed. You can use `g_list_free_full(locations,
 *   g_object_unref)`.
 *
 * Since: 5.0.0
 */
GList *
gaflight_endpoint_get_locations(GAFlightEndpoint *endpoint)
{
  const auto flight_endpoint = gaflight_endpoint_get_raw(endpoint);
  GList *locations = NULL;
  for (const auto &flight_location : flight_endpoint->locations) {
    auto location = gaflight_location_new(flight_location.ToString().c_str(), nullptr);
    locations = g_list_prepend(locations, location);
  }
  return g_list_reverse(locations);
}

typedef struct GAFlightInfoPrivate_
{
  arrow::flight::FlightInfo info;
} GAFlightInfoPrivate;

enum {
  PROP_INFO = 1,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightInfo, gaflight_info, G_TYPE_OBJECT)

#define GAFLIGHT_INFO_GET_PRIVATE(obj)                                                   \
  static_cast<GAFlightInfoPrivate *>(                                                    \
    gaflight_info_get_instance_private(GAFLIGHT_INFO(obj)))

static void
gaflight_info_finalize(GObject *object)
{
  auto priv = GAFLIGHT_INFO_GET_PRIVATE(object);

  priv->info.~FlightInfo();

  G_OBJECT_CLASS(gaflight_info_parent_class)->finalize(object);
}

static void
gaflight_info_set_property(GObject *object,
                           guint prop_id,
                           const GValue *value,
                           GParamSpec *pspec)
{
  auto priv = GAFLIGHT_INFO_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_INFO:
    {
      auto info = static_cast<arrow::flight::FlightInfo *>(g_value_get_pointer(value));
      new (&(priv->info)) arrow::flight::FlightInfo(*info);
    }
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_info_init(GAFlightInfo *object)
{
}

static void
gaflight_info_class_init(GAFlightInfoClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->finalize = gaflight_info_finalize;
  gobject_class->set_property = gaflight_info_set_property;

  GParamSpec *spec;
  spec = g_param_spec_pointer(
    "info",
    "Info",
    "The raw arrow::flight::FlightInfo *",
    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
  g_object_class_install_property(gobject_class, PROP_INFO, spec);
}

/**
 * gaflight_info_new:
 * @schema: A #GArrowSchema.
 * @descriptor: A #GAFlightDescriptor.
 * @endpoints: (element-type GAFlightEndpoint): A list of #GAFlightEndpoint.
 * @total_records: The number of total records.
 * @total_bytes: The number of total bytes.
 * @error: (nullable): Return location for a #GError or %NULL.
 *
 * Returns: (nullable): The newly created #GAFlightInfo, %NULL on error.
 *
 * Since: 5.0.0
 */
GAFlightInfo *
gaflight_info_new(GArrowSchema *schema,
                  GAFlightDescriptor *descriptor,
                  GList *endpoints,
                  gint64 total_records,
                  gint64 total_bytes,
                  GError **error)
{
  auto arrow_schema = garrow_schema_get_raw(schema);
  auto flight_descriptor = gaflight_descriptor_get_raw(descriptor);
  std::vector<arrow::flight::FlightEndpoint> flight_endpoints;
  for (auto node = endpoints; node; node = node->next) {
    auto endpoint = GAFLIGHT_ENDPOINT(node->data);
    flight_endpoints.push_back(*gaflight_endpoint_get_raw(endpoint));
  }
  auto flight_info_result = arrow::flight::FlightInfo::Make(*arrow_schema,
                                                            *flight_descriptor,
                                                            flight_endpoints,
                                                            total_records,
                                                            total_bytes);
  if (!garrow::check(error, flight_info_result, "[flight-info][new]")) {
    return NULL;
  }
  return gaflight_info_new_raw(&(*flight_info_result));
}

/**
 * gaflight_info_equal:
 * @info: A #GAFlightInfo.
 * @other_info: A #GAFlightInfo to be compared.
 *
 * Returns: %TRUE if both of them represents the same information,
 *   %FALSE otherwise.
 *
 * Since: 5.0.0
 */
gboolean
gaflight_info_equal(GAFlightInfo *info, GAFlightInfo *other_info)
{
  const auto flight_info = gaflight_info_get_raw(info);
  const auto flight_other_info = gaflight_info_get_raw(other_info);
  return (flight_info->serialized_schema() == flight_other_info->serialized_schema()) &&
         (flight_info->descriptor() == flight_other_info->descriptor()) &&
         (flight_info->endpoints() == flight_other_info->endpoints()) &&
         (flight_info->total_records() == flight_other_info->total_records()) &&
         (flight_info->total_bytes() == flight_other_info->total_bytes());
}

/**
 * gaflight_info_get_schema:
 * @info: A #GAFlightInfo.
 * @options: (nullable): A #GArrowReadOptions.
 * @error: (nullable): Return location for a #GError or %NULL.
 *
 * Returns: (transfer full): Deserialized #GArrowSchema, %NULL on error.
 *
 * Since: 5.0.0
 */
GArrowSchema *
gaflight_info_get_schema(GAFlightInfo *info, GArrowReadOptions *options, GError **error)
{
  const auto flight_info = gaflight_info_get_raw(info);
  arrow::Status status;
  std::shared_ptr<arrow::Schema> arrow_schema;
  if (options) {
    auto arrow_memo = garrow_read_options_get_dictionary_memo_raw(options);
    status = flight_info->GetSchema(arrow_memo).Value(&arrow_schema);
  } else {
    arrow::ipc::DictionaryMemo arrow_memo;
    status = flight_info->GetSchema(&arrow_memo).Value(&arrow_schema);
  }
  if (garrow::check(error, status, "[flight-info][get-schema]")) {
    return garrow_schema_new_raw(&arrow_schema);
  } else {
    return NULL;
  }
}

/**
 * gaflight_info_get_descriptor:
 * @info: A #GAFlightInfo.
 *
 * Returns: (transfer full): The #GAFlightDescriptor of the information.
 *
 * Since: 5.0.0
 */
GAFlightDescriptor *
gaflight_info_get_descriptor(GAFlightInfo *info)
{
  const auto flight_info = gaflight_info_get_raw(info);
  return gaflight_descriptor_new_raw(&(flight_info->descriptor()));
}

/**
 * gaflight_info_get_endpoints:
 * @info: A #GAFlightInfo.
 *
 * Returns: (element-type GAFlightEndpoint) (transfer full):
 *   The list of #GAFlightEndpoint of the information.
 *
 * Since: 5.0.0
 */
GList *
gaflight_info_get_endpoints(GAFlightInfo *info)
{
  const auto flight_info = gaflight_info_get_raw(info);
  GList *endpoints = NULL;
  for (const auto &flight_endpoint : flight_info->endpoints()) {
    auto endpoint = gaflight_endpoint_new_raw(&flight_endpoint, nullptr);
    endpoints = g_list_prepend(endpoints, endpoint);
  }
  return g_list_reverse(endpoints);
}

/**
 * gaflight_info_get_total_records:
 * @info: A #GAFlightInfo.
 *
 * Returns: The number of total records of the information.
 *
 * Since: 5.0.0
 */
gint64
gaflight_info_get_total_records(GAFlightInfo *info)
{
  const auto flight_info = gaflight_info_get_raw(info);
  return flight_info->total_records();
}

/**
 * gaflight_info_get_total_bytes:
 * @info: A #GAFlightInfo.
 *
 * Returns: The number of total bytes of the information.
 *
 * Since: 5.0.0
 */
gint64
gaflight_info_get_total_bytes(GAFlightInfo *info)
{
  const auto flight_info = gaflight_info_get_raw(info);
  return flight_info->total_bytes();
}

typedef struct GAFlightStreamChunkPrivate_
{
  arrow::flight::FlightStreamChunk chunk;
} GAFlightStreamChunkPrivate;

enum {
  PROP_CHUNK = 1,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightStreamChunk, gaflight_stream_chunk, G_TYPE_OBJECT)

#define GAFLIGHT_STREAM_CHUNK_GET_PRIVATE(obj)                                           \
  static_cast<GAFlightStreamChunkPrivate *>(                                             \
    gaflight_stream_chunk_get_instance_private(GAFLIGHT_STREAM_CHUNK(obj)))

static void
gaflight_stream_chunk_finalize(GObject *object)
{
  auto priv = GAFLIGHT_STREAM_CHUNK_GET_PRIVATE(object);

  priv->chunk.~FlightStreamChunk();

  G_OBJECT_CLASS(gaflight_info_parent_class)->finalize(object);
}

static void
gaflight_stream_chunk_set_property(GObject *object,
                                   guint prop_id,
                                   const GValue *value,
                                   GParamSpec *pspec)
{
  auto priv = GAFLIGHT_STREAM_CHUNK_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_CHUNK:
    priv->chunk =
      *static_cast<arrow::flight::FlightStreamChunk *>(g_value_get_pointer(value));
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_stream_chunk_init(GAFlightStreamChunk *object)
{
}

static void
gaflight_stream_chunk_class_init(GAFlightStreamChunkClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->finalize = gaflight_stream_chunk_finalize;
  gobject_class->set_property = gaflight_stream_chunk_set_property;

  GParamSpec *spec;
  spec = g_param_spec_pointer(
    "chunk",
    "Stream chunk",
    "The raw arrow::flight::FlightStreamChunk *",
    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
  g_object_class_install_property(gobject_class, PROP_CHUNK, spec);
}

/**
 * gaflight_stream_chunk_get_data:
 * @chunk: A #GAFlightStreamChunk.
 *
 * Returns: (transfer full): The data of the chunk.
 *
 * Since: 6.0.0
 */
GArrowRecordBatch *
gaflight_stream_chunk_get_data(GAFlightStreamChunk *chunk)
{
  auto flight_chunk = gaflight_stream_chunk_get_raw(chunk);
  return garrow_record_batch_new_raw(&(flight_chunk->data));
}

/**
 * gaflight_stream_chunk_get_metadata:
 * @chunk: A #GAFlightStreamChunk.
 *
 * Returns: (nullable) (transfer full): The metadata of the chunk.
 *
 *   The metadata may be NULL.
 *
 * Since: 6.0.0
 */
GArrowBuffer *
gaflight_stream_chunk_get_metadata(GAFlightStreamChunk *chunk)
{
  auto flight_chunk = gaflight_stream_chunk_get_raw(chunk);
  if (flight_chunk->app_metadata) {
    return garrow_buffer_new_raw(&(flight_chunk->app_metadata));
  } else {
    return NULL;
  }
}

typedef struct GAFlightRecordBatchReaderPrivate_
{
  arrow::flight::MetadataRecordBatchReader *reader;
  bool is_owner;
} GAFlightRecordBatchReaderPrivate;

enum {
  PROP_READER = 1,
  PROP_IS_OWNER,
};

G_DEFINE_TYPE_WITH_PRIVATE(GAFlightRecordBatchReader,
                           gaflight_record_batch_reader,
                           G_TYPE_OBJECT)

#define GAFLIGHT_RECORD_BATCH_READER_GET_PRIVATE(obj)                                    \
  static_cast<GAFlightRecordBatchReaderPrivate *>(                                       \
    gaflight_record_batch_reader_get_instance_private(                                   \
      GAFLIGHT_RECORD_BATCH_READER(obj)))

static void
gaflight_record_batch_reader_finalize(GObject *object)
{
  auto priv = GAFLIGHT_RECORD_BATCH_READER_GET_PRIVATE(object);
  if (priv->is_owner) {
    delete priv->reader;
  }
  G_OBJECT_CLASS(gaflight_info_parent_class)->finalize(object);
}

static void
gaflight_record_batch_reader_set_property(GObject *object,
                                          guint prop_id,
                                          const GValue *value,
                                          GParamSpec *pspec)
{
  auto priv = GAFLIGHT_RECORD_BATCH_READER_GET_PRIVATE(object);

  switch (prop_id) {
  case PROP_READER:
    priv->reader =
      static_cast<arrow::flight::MetadataRecordBatchReader *>(g_value_get_pointer(value));
    break;
  case PROP_IS_OWNER:
    priv->is_owner = g_value_get_boolean(value);
    break;
  default:
    G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec);
    break;
  }
}

static void
gaflight_record_batch_reader_init(GAFlightRecordBatchReader *object)
{
}

static void
gaflight_record_batch_reader_class_init(GAFlightRecordBatchReaderClass *klass)
{
  auto gobject_class = G_OBJECT_CLASS(klass);

  gobject_class->finalize = gaflight_record_batch_reader_finalize;
  gobject_class->set_property = gaflight_record_batch_reader_set_property;

  GParamSpec *spec;
  spec = g_param_spec_pointer(
    "reader",
    nullptr,
    nullptr,
    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
  g_object_class_install_property(gobject_class, PROP_READER, spec);

  spec = g_param_spec_boolean(
    "is-owner",
    nullptr,
    nullptr,
    TRUE,
    static_cast<GParamFlags>(G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
  g_object_class_install_property(gobject_class, PROP_IS_OWNER, spec);
}

/**
 * gaflight_record_batch_reader_read_next:
 * @reader: A #GAFlightRecordBatchReader.
 * @error: (nullable): Return location for a #GError or %NULL.
 *
 * Returns: (transfer full): The next chunk on success, %NULL on end
 *   of stream, %NULL on error.
 *
 * Since: 6.0.0
 */
GAFlightStreamChunk *
gaflight_record_batch_reader_read_next(GAFlightRecordBatchReader *reader, GError **error)
{
  auto flight_reader = gaflight_record_batch_reader_get_raw(reader);
  arrow::flight::FlightStreamChunk flight_chunk;
  auto status = flight_reader->Next().Value(&flight_chunk);
  if (garrow::check(error, status, "[flight-record-batch-reader][read-next]")) {
    if (flight_chunk.data) {
      return gaflight_stream_chunk_new_raw(&flight_chunk);
    } else {
      return NULL;
    }
  } else {
    return NULL;
  }
}

/**
 * gaflight_record_batch_reader_read_all:
 * @reader: A #GAFlightRecordBatchReader.
 * @error: (nullable): Return location for a #GError or %NULL.
 *
 * Returns: (transfer full): The all data on success, %NULL on error.
 *
 * Since: 6.0.0
 */
GArrowTable *
gaflight_record_batch_reader_read_all(GAFlightRecordBatchReader *reader, GError **error)
{
  auto flight_reader = gaflight_record_batch_reader_get_raw(reader);
  std::shared_ptr<arrow::Table> arrow_table;
  auto status = flight_reader->ToTable().Value(&arrow_table);
  if (garrow::check(error, status, "[flight-record-batch-reader][read-all]")) {
    return garrow_table_new_raw(&arrow_table);
  } else {
    return NULL;
  }
}

G_END_DECLS

GAFlightCriteria *
gaflight_criteria_new_raw(const arrow::flight::Criteria *flight_criteria)
{
  auto criteria = g_object_new(GAFLIGHT_TYPE_CRITERIA, NULL);
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(criteria);
  priv->criteria = *flight_criteria;
  priv->expression =
    g_bytes_new(priv->criteria.expression.data(), priv->criteria.expression.size());
  return GAFLIGHT_CRITERIA(criteria);
}

arrow::flight::Criteria *
gaflight_criteria_get_raw(GAFlightCriteria *criteria)
{
  auto priv = GAFLIGHT_CRITERIA_GET_PRIVATE(criteria);
  return &(priv->criteria);
}

arrow::flight::Location *
gaflight_location_get_raw(GAFlightLocation *location)
{
  auto priv = GAFLIGHT_LOCATION_GET_PRIVATE(location);
  return &(priv->location);
}

GAFlightDescriptor *
gaflight_descriptor_new_raw(const arrow::flight::FlightDescriptor *flight_descriptor)
{
  GType gtype = GAFLIGHT_TYPE_DESCRIPTOR;
  switch (flight_descriptor->type) {
  case arrow::flight::FlightDescriptor::DescriptorType::PATH:
    gtype = GAFLIGHT_TYPE_PATH_DESCRIPTOR;
    break;
  case arrow::flight::FlightDescriptor::DescriptorType::CMD:
    gtype = GAFLIGHT_TYPE_COMMAND_DESCRIPTOR;
    break;
  default:
    break;
  }
  return GAFLIGHT_DESCRIPTOR(g_object_new(gtype, "descriptor", flight_descriptor, NULL));
}

arrow::flight::FlightDescriptor *
gaflight_descriptor_get_raw(GAFlightDescriptor *descriptor)
{
  auto priv = GAFLIGHT_DESCRIPTOR_GET_PRIVATE(descriptor);
  return &(priv->descriptor);
}

GAFlightTicket *
gaflight_ticket_new_raw(const arrow::flight::Ticket *flight_ticket)
{
  auto ticket = g_object_new(GAFLIGHT_TYPE_TICKET, NULL);
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(ticket);
  priv->ticket = *flight_ticket;
  priv->data = g_bytes_new(priv->ticket.ticket.data(), priv->ticket.ticket.size());
  return GAFLIGHT_TICKET(ticket);
}

arrow::flight::Ticket *
gaflight_ticket_get_raw(GAFlightTicket *ticket)
{
  auto priv = GAFLIGHT_TICKET_GET_PRIVATE(ticket);
  return &(priv->ticket);
}

GAFlightEndpoint *
gaflight_endpoint_new_raw(const arrow::flight::FlightEndpoint *flight_endpoint,
                          GAFlightTicket *ticket)
{
  auto endpoint = GAFLIGHT_ENDPOINT(g_object_new(GAFLIGHT_TYPE_ENDPOINT, NULL));
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(endpoint);
  if (ticket) {
    priv->ticket = ticket;
    g_object_ref(priv->ticket);
    priv->endpoint.ticket = *gaflight_ticket_get_raw(priv->ticket);
  } else {
    auto data = g_bytes_new(flight_endpoint->ticket.ticket.data(),
                            flight_endpoint->ticket.ticket.length());
    auto ticket = gaflight_ticket_new(data);
    g_bytes_unref(data);
    priv->ticket = ticket;
    priv->endpoint.ticket.ticket = flight_endpoint->ticket.ticket;
  }
  if (flight_endpoint) {
    priv->endpoint.locations = flight_endpoint->locations;
  }
  return endpoint;
}

arrow::flight::FlightEndpoint *
gaflight_endpoint_get_raw(GAFlightEndpoint *endpoint)
{
  auto priv = GAFLIGHT_ENDPOINT_GET_PRIVATE(endpoint);
  return &(priv->endpoint);
}

GAFlightInfo *
gaflight_info_new_raw(arrow::flight::FlightInfo *flight_info)
{
  return GAFLIGHT_INFO(g_object_new(GAFLIGHT_TYPE_INFO, "info", flight_info, NULL));
}

arrow::flight::FlightInfo *
gaflight_info_get_raw(GAFlightInfo *info)
{
  auto priv = GAFLIGHT_INFO_GET_PRIVATE(info);
  return &(priv->info);
}

GAFlightStreamChunk *
gaflight_stream_chunk_new_raw(arrow::flight::FlightStreamChunk *flight_chunk)
{
  return GAFLIGHT_STREAM_CHUNK(
    g_object_new(GAFLIGHT_TYPE_STREAM_CHUNK, "chunk", flight_chunk, NULL));
}

arrow::flight::FlightStreamChunk *
gaflight_stream_chunk_get_raw(GAFlightStreamChunk *chunk)
{
  auto priv = GAFLIGHT_STREAM_CHUNK_GET_PRIVATE(chunk);
  return &(priv->chunk);
}

arrow::flight::MetadataRecordBatchReader *
gaflight_record_batch_reader_get_raw(GAFlightRecordBatchReader *reader)
{
  auto priv = GAFLIGHT_RECORD_BATCH_READER_GET_PRIVATE(reader);
  return priv->reader;
}