/**
 * Autogenerated by Thrift Compiler (0.9.2)
 *
 * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
 *  @generated
 */
#include <string.h>
#include <thrift/c_glib/thrift.h>
#include <thrift/c_glib/thrift_application_exception.h>
#include "consumer_service.h"

gboolean
consumer_service_if_pull (ConsumerServiceIf *iface, PullResponse ** _return, const PullRequest * request, PullException ** error, GError **error)
{
  return CONSUMER_SERVICE_IF_GET_INTERFACE (iface)->pull (iface, _return, request, error, error);
}

gboolean
consumer_service_if_submit (ConsumerServiceIf *iface, gboolean* _return, const ConsumeResult * result, PullException ** error, GError **error)
{
  return CONSUMER_SERVICE_IF_GET_INTERFACE (iface)->submit (iface, _return, result, error, error);
}

gboolean
consumer_service_if_get_consume_stats (ConsumerServiceIf *iface, GPtrArray ** _return, const ConsumeStatsRequest * request, PullException ** error, GError **error)
{
  return CONSUMER_SERVICE_IF_GET_INTERFACE (iface)->get_consume_stats (iface, _return, request, error, error);
}

gboolean
consumer_service_if_fetch (ConsumerServiceIf *iface, FetchResponse ** _return, const FetchRequest * request, GError **error)
{
  return CONSUMER_SERVICE_IF_GET_INTERFACE (iface)->fetch (iface, _return, request, error);
}

gboolean
consumer_service_if_ack (ConsumerServiceIf *iface, gboolean* _return, const AckResult * result, GError **error)
{
  return CONSUMER_SERVICE_IF_GET_INTERFACE (iface)->ack (iface, _return, result, error);
}

GType
consumer_service_if_get_type (void)
{
  static GType type = 0;
  if (type == 0)
  {
    static const GTypeInfo type_info =
    {
      sizeof (ConsumerServiceIfInterface),
      NULL,  /* base_init */
      NULL,  /* base_finalize */
      NULL,  /* class_init */
      NULL,  /* class_finalize */
      NULL,  /* class_data */
      0,     /* instance_size */
      0,     /* n_preallocs */
      NULL,  /* instance_init */
      NULL   /* value_table */
    };
    type = g_type_register_static (G_TYPE_INTERFACE,
                                   "ConsumerServiceIf",
                                   &type_info, 0);
  }
  return type;
}

static void 
consumer_service_if_interface_init (ConsumerServiceIfInterface *iface);

G_DEFINE_TYPE_WITH_CODE (ConsumerServiceClient, consumer_service_client,
                         G_TYPE_OBJECT, 
                         G_IMPLEMENT_INTERFACE (TYPE_CONSUMER_SERVICE_IF,
                                                consumer_service_if_interface_init));

enum _ConsumerServiceClientProperties
{
  PROP_0,
  PROP_CONSUMER_SERVICE_CLIENT_INPUT_PROTOCOL,
  PROP_CONSUMER_SERVICE_CLIENT_OUTPUT_PROTOCOL
};

void
consumer_service_client_set_property (GObject *object, guint property_id, const GValue *value, GParamSpec *pspec)
{
  ConsumerServiceClient *client = CONSUMER_SERVICE_CLIENT (object);

  THRIFT_UNUSED_VAR (pspec);

  switch (property_id)
  {
    case PROP_CONSUMER_SERVICE_CLIENT_INPUT_PROTOCOL:
      client->input_protocol = g_value_get_object (value);
      break;
    case PROP_CONSUMER_SERVICE_CLIENT_OUTPUT_PROTOCOL:
      client->output_protocol = g_value_get_object (value);
      break;
  }
}

void
consumer_service_client_get_property (GObject *object, guint property_id, GValue *value, GParamSpec *pspec)
{
  ConsumerServiceClient *client = CONSUMER_SERVICE_CLIENT (object);

  THRIFT_UNUSED_VAR (pspec);

  switch (property_id)
  {
    case PROP_CONSUMER_SERVICE_CLIENT_INPUT_PROTOCOL:
      g_value_set_object (value, client->input_protocol);
      break;
    case PROP_CONSUMER_SERVICE_CLIENT_OUTPUT_PROTOCOL:
      g_value_set_object (value, client->output_protocol);
      break;
  }
}

gboolean consumer_service_client_send_pull (ConsumerServiceIf * iface, const PullRequest * request, GError ** error)
{
  gint32 cseqid = 0;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->output_protocol;

  if (thrift_protocol_write_message_begin (protocol, "pull", T_CALL, cseqid, error) < 0)
    return FALSE;

  {
    gint32 ret;
    gint32 xfer = 0;

    
    if ((ret = thrift_protocol_write_struct_begin (protocol, "pull_args", error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_begin (protocol, "request", T_STRUCT, 1, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_struct_write (THRIFT_STRUCT (request), protocol, error)) < 0)
      return 0;
    xfer += ret;

    if ((ret = thrift_protocol_write_field_end (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_stop (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_write_message_end (protocol, error) < 0)
    return FALSE;
  if (!thrift_transport_flush (protocol->transport, error))
    return FALSE;
  if (!thrift_transport_write_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_recv_pull (ConsumerServiceIf * iface, PullResponse ** _return, PullException ** error, GError ** error)
{

  gint32 rseqid;
  gchar * fname = NULL;
  ThriftMessageType mtype;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->input_protocol;

  if (thrift_protocol_read_message_begin (protocol, &fname, &mtype, &rseqid, error) < 0)
  {
    if (fname) g_free (fname);
    return FALSE;
  }

  if (mtype == T_EXCEPTION) {
    if (fname) g_free (fname);
    ThriftApplicationException *xception = g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION, NULL);
    thrift_struct_read (THRIFT_STRUCT (xception), protocol, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, xception->type, "application error: %s", xception->message);
    g_object_unref (xception);
    return FALSE;
  } else if (mtype != T_REPLY) {
    if (fname) g_free (fname);
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_INVALID_MESSAGE_TYPE, "invalid message type %d, expected T_REPLY", mtype);
    return FALSE;
  } else if (strncmp (fname, "pull", 4) != 0) {
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, error);
    thrift_transport_read_end (protocol->transport, error);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_WRONG_METHOD_NAME, "wrong method name %s, expected pull", fname);
    if (fname) g_free (fname);
    return FALSE;
  }
  if (fname) g_free (fname);

  {
    gint32 ret;
    gint32 xfer = 0;
    gchar *name = NULL;
    ThriftType ftype;
    gint16 fid;
    guint32 len = 0;
    gpointer data = NULL;
    

    /* satisfy -Wall in case these aren't used */
    THRIFT_UNUSED_VAR (len);
    THRIFT_UNUSED_VAR (data);

    /* read the struct begin marker */
    if ((ret = thrift_protocol_read_struct_begin (protocol, &name, error)) < 0)
    {
      if (name) g_free (name);
      return 0;
    }
    xfer += ret;
    if (name) g_free (name);
    name = NULL;

    /* read the struct fields */
    while (1)
    {
      /* read the beginning of a field */
      if ((ret = thrift_protocol_read_field_begin (protocol, &name, &ftype, &fid, error)) < 0)
      {
        if (name) g_free (name);
        return 0;
      }
      xfer += ret;
      if (name) g_free (name);
      name = NULL;

      /* break if we get a STOP field */
      if (ftype == T_STOP)
      {
        break;
      }

      switch (fid)
      {
        case 0:
          if (ftype == T_STRUCT)
          {
            if ((ret = thrift_struct_read (THRIFT_STRUCT (*_return), protocol, error)) < 0)
            {
              return 0;
            }
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        case 1:
          if (ftype == T_STRUCT)
          {
            /* This struct is an exception */
            if ( *error != NULL)
            {
              g_object_unref (*error);
            }
            *error = g_object_new (TYPE_PULL_EXCEPTION, NULL);
            if ((ret = thrift_struct_read (THRIFT_STRUCT (*error), protocol, error)) < 0)
            {
              g_object_unref (*error);
              *error = NULL;
              return 0;
            }
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        default:
          if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
            return 0;
          xfer += ret;
          break;
      }
      if ((ret = thrift_protocol_read_field_end (protocol, error)) < 0)
        return 0;
      xfer += ret;
    }

    if ((ret = thrift_protocol_read_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_read_message_end (protocol, error) < 0)
    return FALSE;

  if (!thrift_transport_read_end (protocol->transport, error))
    return FALSE;

  if (*error != NULL)
  {
      g_set_error (error, PULL_EXCEPTION_ERROR, PULL_EXCEPTION_ERROR_CODE, "PullException");
      return FALSE;
  }
  return TRUE;
}

gboolean consumer_service_client_pull (ConsumerServiceIf * iface, PullResponse ** _return, const PullRequest * request, PullException ** error, GError ** error)
{
  if (!consumer_service_client_send_pull (iface, request, error))
    return FALSE;
  if (!consumer_service_client_recv_pull (iface, _return, error, error))
    return FALSE;
  return TRUE;
}

gboolean consumer_service_client_send_submit (ConsumerServiceIf * iface, const ConsumeResult * result, GError ** error)
{
  gint32 cseqid = 0;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->output_protocol;

  if (thrift_protocol_write_message_begin (protocol, "submit", T_CALL, cseqid, error) < 0)
    return FALSE;

  {
    gint32 ret;
    gint32 xfer = 0;

    
    if ((ret = thrift_protocol_write_struct_begin (protocol, "submit_args", error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_begin (protocol, "result", T_STRUCT, 1, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_struct_write (THRIFT_STRUCT (result), protocol, error)) < 0)
      return 0;
    xfer += ret;

    if ((ret = thrift_protocol_write_field_end (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_stop (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_write_message_end (protocol, error) < 0)
    return FALSE;
  if (!thrift_transport_flush (protocol->transport, error))
    return FALSE;
  if (!thrift_transport_write_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_recv_submit (ConsumerServiceIf * iface, gboolean* _return, PullException ** error, GError ** error)
{

  gint32 rseqid;
  gchar * fname = NULL;
  ThriftMessageType mtype;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->input_protocol;

  if (thrift_protocol_read_message_begin (protocol, &fname, &mtype, &rseqid, error) < 0)
  {
    if (fname) g_free (fname);
    return FALSE;
  }

  if (mtype == T_EXCEPTION) {
    if (fname) g_free (fname);
    ThriftApplicationException *xception = g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION, NULL);
    thrift_struct_read (THRIFT_STRUCT (xception), protocol, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, xception->type, "application error: %s", xception->message);
    g_object_unref (xception);
    return FALSE;
  } else if (mtype != T_REPLY) {
    if (fname) g_free (fname);
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_INVALID_MESSAGE_TYPE, "invalid message type %d, expected T_REPLY", mtype);
    return FALSE;
  } else if (strncmp (fname, "submit", 6) != 0) {
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, error);
    thrift_transport_read_end (protocol->transport, error);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_WRONG_METHOD_NAME, "wrong method name %s, expected submit", fname);
    if (fname) g_free (fname);
    return FALSE;
  }
  if (fname) g_free (fname);

  {
    gint32 ret;
    gint32 xfer = 0;
    gchar *name = NULL;
    ThriftType ftype;
    gint16 fid;
    guint32 len = 0;
    gpointer data = NULL;
    

    /* satisfy -Wall in case these aren't used */
    THRIFT_UNUSED_VAR (len);
    THRIFT_UNUSED_VAR (data);

    /* read the struct begin marker */
    if ((ret = thrift_protocol_read_struct_begin (protocol, &name, error)) < 0)
    {
      if (name) g_free (name);
      return 0;
    }
    xfer += ret;
    if (name) g_free (name);
    name = NULL;

    /* read the struct fields */
    while (1)
    {
      /* read the beginning of a field */
      if ((ret = thrift_protocol_read_field_begin (protocol, &name, &ftype, &fid, error)) < 0)
      {
        if (name) g_free (name);
        return 0;
      }
      xfer += ret;
      if (name) g_free (name);
      name = NULL;

      /* break if we get a STOP field */
      if (ftype == T_STOP)
      {
        break;
      }

      switch (fid)
      {
        case 0:
          if (ftype == T_BOOL)
          {
            if ((ret = thrift_protocol_read_bool (protocol, &*_return, error)) < 0)
              return 0;
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        case 1:
          if (ftype == T_STRUCT)
          {
            /* This struct is an exception */
            if ( *error != NULL)
            {
              g_object_unref (*error);
            }
            *error = g_object_new (TYPE_PULL_EXCEPTION, NULL);
            if ((ret = thrift_struct_read (THRIFT_STRUCT (*error), protocol, error)) < 0)
            {
              g_object_unref (*error);
              *error = NULL;
              return 0;
            }
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        default:
          if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
            return 0;
          xfer += ret;
          break;
      }
      if ((ret = thrift_protocol_read_field_end (protocol, error)) < 0)
        return 0;
      xfer += ret;
    }

    if ((ret = thrift_protocol_read_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_read_message_end (protocol, error) < 0)
    return FALSE;

  if (!thrift_transport_read_end (protocol->transport, error))
    return FALSE;

  if (*error != NULL)
  {
      g_set_error (error, PULL_EXCEPTION_ERROR, PULL_EXCEPTION_ERROR_CODE, "PullException");
      return FALSE;
  }
  return TRUE;
}

gboolean consumer_service_client_submit (ConsumerServiceIf * iface, gboolean* _return, const ConsumeResult * result, PullException ** error, GError ** error)
{
  if (!consumer_service_client_send_submit (iface, result, error))
    return FALSE;
  if (!consumer_service_client_recv_submit (iface, _return, error, error))
    return FALSE;
  return TRUE;
}

gboolean consumer_service_client_send_get_consume_stats (ConsumerServiceIf * iface, const ConsumeStatsRequest * request, GError ** error)
{
  gint32 cseqid = 0;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->output_protocol;

  if (thrift_protocol_write_message_begin (protocol, "getConsumeStats", T_CALL, cseqid, error) < 0)
    return FALSE;

  {
    gint32 ret;
    gint32 xfer = 0;

    
    if ((ret = thrift_protocol_write_struct_begin (protocol, "getConsumeStats_args", error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_begin (protocol, "request", T_STRUCT, 1, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_struct_write (THRIFT_STRUCT (request), protocol, error)) < 0)
      return 0;
    xfer += ret;

    if ((ret = thrift_protocol_write_field_end (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_stop (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_write_message_end (protocol, error) < 0)
    return FALSE;
  if (!thrift_transport_flush (protocol->transport, error))
    return FALSE;
  if (!thrift_transport_write_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_recv_get_consume_stats (ConsumerServiceIf * iface, GPtrArray ** _return, PullException ** error, GError ** error)
{

  gint32 rseqid;
  gchar * fname = NULL;
  ThriftMessageType mtype;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->input_protocol;

  if (thrift_protocol_read_message_begin (protocol, &fname, &mtype, &rseqid, error) < 0)
  {
    if (fname) g_free (fname);
    return FALSE;
  }

  if (mtype == T_EXCEPTION) {
    if (fname) g_free (fname);
    ThriftApplicationException *xception = g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION, NULL);
    thrift_struct_read (THRIFT_STRUCT (xception), protocol, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, xception->type, "application error: %s", xception->message);
    g_object_unref (xception);
    return FALSE;
  } else if (mtype != T_REPLY) {
    if (fname) g_free (fname);
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_INVALID_MESSAGE_TYPE, "invalid message type %d, expected T_REPLY", mtype);
    return FALSE;
  } else if (strncmp (fname, "getConsumeStats", 15) != 0) {
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, error);
    thrift_transport_read_end (protocol->transport, error);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_WRONG_METHOD_NAME, "wrong method name %s, expected getConsumeStats", fname);
    if (fname) g_free (fname);
    return FALSE;
  }
  if (fname) g_free (fname);

  {
    gint32 ret;
    gint32 xfer = 0;
    gchar *name = NULL;
    ThriftType ftype;
    gint16 fid;
    guint32 len = 0;
    gpointer data = NULL;
    

    /* satisfy -Wall in case these aren't used */
    THRIFT_UNUSED_VAR (len);
    THRIFT_UNUSED_VAR (data);

    /* read the struct begin marker */
    if ((ret = thrift_protocol_read_struct_begin (protocol, &name, error)) < 0)
    {
      if (name) g_free (name);
      return 0;
    }
    xfer += ret;
    if (name) g_free (name);
    name = NULL;

    /* read the struct fields */
    while (1)
    {
      /* read the beginning of a field */
      if ((ret = thrift_protocol_read_field_begin (protocol, &name, &ftype, &fid, error)) < 0)
      {
        if (name) g_free (name);
        return 0;
      }
      xfer += ret;
      if (name) g_free (name);
      name = NULL;

      /* break if we get a STOP field */
      if (ftype == T_STOP)
      {
        break;
      }

      switch (fid)
      {
        case 0:
          if (ftype == T_LIST)
          {
            {
              guint32 size;
              ThriftType element_type;
              if ((ret = thrift_protocol_read_list_begin (protocol, &element_type, &size, error)) < 0)
                return 0;
              xfer += ret;

              /* iterate through list elements */
              guint32 i;
              for (i = 0; i < size; i++)
              {
                ConsumeStats * _elem36 = NULL;
                if ( _elem36 != NULL)
                {
                  g_object_unref (_elem36);
                }
                _elem36 = g_object_new (TYPE_CONSUME_STATS, NULL);
                if ((ret = thrift_struct_read (THRIFT_STRUCT (_elem36), protocol, error)) < 0)
                {
                  g_object_unref (_elem36);
                  return 0;
                }
                xfer += ret;
                g_ptr_array_add (*_return, _elem36);
              }
              if ((ret = thrift_protocol_read_list_end (protocol, error)) < 0)
                return 0;
              xfer += ret;

            }
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        case 1:
          if (ftype == T_STRUCT)
          {
            /* This struct is an exception */
            if ( *error != NULL)
            {
              g_object_unref (*error);
            }
            *error = g_object_new (TYPE_PULL_EXCEPTION, NULL);
            if ((ret = thrift_struct_read (THRIFT_STRUCT (*error), protocol, error)) < 0)
            {
              g_object_unref (*error);
              *error = NULL;
              return 0;
            }
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        default:
          if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
            return 0;
          xfer += ret;
          break;
      }
      if ((ret = thrift_protocol_read_field_end (protocol, error)) < 0)
        return 0;
      xfer += ret;
    }

    if ((ret = thrift_protocol_read_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_read_message_end (protocol, error) < 0)
    return FALSE;

  if (!thrift_transport_read_end (protocol->transport, error))
    return FALSE;

  if (*error != NULL)
  {
      g_set_error (error, PULL_EXCEPTION_ERROR, PULL_EXCEPTION_ERROR_CODE, "PullException");
      return FALSE;
  }
  return TRUE;
}

gboolean consumer_service_client_get_consume_stats (ConsumerServiceIf * iface, GPtrArray ** _return, const ConsumeStatsRequest * request, PullException ** error, GError ** error)
{
  if (!consumer_service_client_send_get_consume_stats (iface, request, error))
    return FALSE;
  if (!consumer_service_client_recv_get_consume_stats (iface, _return, error, error))
    return FALSE;
  return TRUE;
}

gboolean consumer_service_client_send_fetch (ConsumerServiceIf * iface, const FetchRequest * request, GError ** error)
{
  gint32 cseqid = 0;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->output_protocol;

  if (thrift_protocol_write_message_begin (protocol, "fetch", T_CALL, cseqid, error) < 0)
    return FALSE;

  {
    gint32 ret;
    gint32 xfer = 0;

    
    if ((ret = thrift_protocol_write_struct_begin (protocol, "fetch_args", error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_begin (protocol, "request", T_STRUCT, 1, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_struct_write (THRIFT_STRUCT (request), protocol, error)) < 0)
      return 0;
    xfer += ret;

    if ((ret = thrift_protocol_write_field_end (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_stop (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_write_message_end (protocol, error) < 0)
    return FALSE;
  if (!thrift_transport_flush (protocol->transport, error))
    return FALSE;
  if (!thrift_transport_write_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_recv_fetch (ConsumerServiceIf * iface, FetchResponse ** _return, GError ** error)
{

  gint32 rseqid;
  gchar * fname = NULL;
  ThriftMessageType mtype;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->input_protocol;

  if (thrift_protocol_read_message_begin (protocol, &fname, &mtype, &rseqid, error) < 0)
  {
    if (fname) g_free (fname);
    return FALSE;
  }

  if (mtype == T_EXCEPTION) {
    if (fname) g_free (fname);
    ThriftApplicationException *xception = g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION, NULL);
    thrift_struct_read (THRIFT_STRUCT (xception), protocol, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, xception->type, "application error: %s", xception->message);
    g_object_unref (xception);
    return FALSE;
  } else if (mtype != T_REPLY) {
    if (fname) g_free (fname);
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_INVALID_MESSAGE_TYPE, "invalid message type %d, expected T_REPLY", mtype);
    return FALSE;
  } else if (strncmp (fname, "fetch", 5) != 0) {
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, error);
    thrift_transport_read_end (protocol->transport, error);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_WRONG_METHOD_NAME, "wrong method name %s, expected fetch", fname);
    if (fname) g_free (fname);
    return FALSE;
  }
  if (fname) g_free (fname);

  {
    gint32 ret;
    gint32 xfer = 0;
    gchar *name = NULL;
    ThriftType ftype;
    gint16 fid;
    guint32 len = 0;
    gpointer data = NULL;
    

    /* satisfy -Wall in case these aren't used */
    THRIFT_UNUSED_VAR (len);
    THRIFT_UNUSED_VAR (data);

    /* read the struct begin marker */
    if ((ret = thrift_protocol_read_struct_begin (protocol, &name, error)) < 0)
    {
      if (name) g_free (name);
      return 0;
    }
    xfer += ret;
    if (name) g_free (name);
    name = NULL;

    /* read the struct fields */
    while (1)
    {
      /* read the beginning of a field */
      if ((ret = thrift_protocol_read_field_begin (protocol, &name, &ftype, &fid, error)) < 0)
      {
        if (name) g_free (name);
        return 0;
      }
      xfer += ret;
      if (name) g_free (name);
      name = NULL;

      /* break if we get a STOP field */
      if (ftype == T_STOP)
      {
        break;
      }

      switch (fid)
      {
        case 0:
          if (ftype == T_STRUCT)
          {
            if ((ret = thrift_struct_read (THRIFT_STRUCT (*_return), protocol, error)) < 0)
            {
              return 0;
            }
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        default:
          if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
            return 0;
          xfer += ret;
          break;
      }
      if ((ret = thrift_protocol_read_field_end (protocol, error)) < 0)
        return 0;
      xfer += ret;
    }

    if ((ret = thrift_protocol_read_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_read_message_end (protocol, error) < 0)
    return FALSE;

  if (!thrift_transport_read_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_fetch (ConsumerServiceIf * iface, FetchResponse ** _return, const FetchRequest * request, GError ** error)
{
  if (!consumer_service_client_send_fetch (iface, request, error))
    return FALSE;
  if (!consumer_service_client_recv_fetch (iface, _return, error))
    return FALSE;
  return TRUE;
}

gboolean consumer_service_client_send_ack (ConsumerServiceIf * iface, const AckResult * result, GError ** error)
{
  gint32 cseqid = 0;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->output_protocol;

  if (thrift_protocol_write_message_begin (protocol, "ack", T_CALL, cseqid, error) < 0)
    return FALSE;

  {
    gint32 ret;
    gint32 xfer = 0;

    
    if ((ret = thrift_protocol_write_struct_begin (protocol, "ack_args", error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_begin (protocol, "result", T_STRUCT, 1, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_struct_write (THRIFT_STRUCT (result), protocol, error)) < 0)
      return 0;
    xfer += ret;

    if ((ret = thrift_protocol_write_field_end (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_field_stop (protocol, error)) < 0)
      return 0;
    xfer += ret;
    if ((ret = thrift_protocol_write_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_write_message_end (protocol, error) < 0)
    return FALSE;
  if (!thrift_transport_flush (protocol->transport, error))
    return FALSE;
  if (!thrift_transport_write_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_recv_ack (ConsumerServiceIf * iface, gboolean* _return, GError ** error)
{

  gint32 rseqid;
  gchar * fname = NULL;
  ThriftMessageType mtype;
  ThriftProtocol * protocol = CONSUMER_SERVICE_CLIENT (iface)->input_protocol;

  if (thrift_protocol_read_message_begin (protocol, &fname, &mtype, &rseqid, error) < 0)
  {
    if (fname) g_free (fname);
    return FALSE;
  }

  if (mtype == T_EXCEPTION) {
    if (fname) g_free (fname);
    ThriftApplicationException *xception = g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION, NULL);
    thrift_struct_read (THRIFT_STRUCT (xception), protocol, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, xception->type, "application error: %s", xception->message);
    g_object_unref (xception);
    return FALSE;
  } else if (mtype != T_REPLY) {
    if (fname) g_free (fname);
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, NULL);
    thrift_transport_read_end (protocol->transport, NULL);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_INVALID_MESSAGE_TYPE, "invalid message type %d, expected T_REPLY", mtype);
    return FALSE;
  } else if (strncmp (fname, "ack", 3) != 0) {
    thrift_protocol_skip (protocol, T_STRUCT, NULL);
    thrift_protocol_read_message_end (protocol, error);
    thrift_transport_read_end (protocol->transport, error);
    g_set_error (error, THRIFT_APPLICATION_EXCEPTION_ERROR, THRIFT_APPLICATION_EXCEPTION_ERROR_WRONG_METHOD_NAME, "wrong method name %s, expected ack", fname);
    if (fname) g_free (fname);
    return FALSE;
  }
  if (fname) g_free (fname);

  {
    gint32 ret;
    gint32 xfer = 0;
    gchar *name = NULL;
    ThriftType ftype;
    gint16 fid;
    guint32 len = 0;
    gpointer data = NULL;
    

    /* satisfy -Wall in case these aren't used */
    THRIFT_UNUSED_VAR (len);
    THRIFT_UNUSED_VAR (data);

    /* read the struct begin marker */
    if ((ret = thrift_protocol_read_struct_begin (protocol, &name, error)) < 0)
    {
      if (name) g_free (name);
      return 0;
    }
    xfer += ret;
    if (name) g_free (name);
    name = NULL;

    /* read the struct fields */
    while (1)
    {
      /* read the beginning of a field */
      if ((ret = thrift_protocol_read_field_begin (protocol, &name, &ftype, &fid, error)) < 0)
      {
        if (name) g_free (name);
        return 0;
      }
      xfer += ret;
      if (name) g_free (name);
      name = NULL;

      /* break if we get a STOP field */
      if (ftype == T_STOP)
      {
        break;
      }

      switch (fid)
      {
        case 0:
          if (ftype == T_BOOL)
          {
            if ((ret = thrift_protocol_read_bool (protocol, &*_return, error)) < 0)
              return 0;
            xfer += ret;
          } else {
            if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
              return 0;
            xfer += ret;
          }
          break;
        default:
          if ((ret = thrift_protocol_skip (protocol, ftype, error)) < 0)
            return 0;
          xfer += ret;
          break;
      }
      if ((ret = thrift_protocol_read_field_end (protocol, error)) < 0)
        return 0;
      xfer += ret;
    }

    if ((ret = thrift_protocol_read_struct_end (protocol, error)) < 0)
      return 0;
    xfer += ret;

  }

  if (thrift_protocol_read_message_end (protocol, error) < 0)
    return FALSE;

  if (!thrift_transport_read_end (protocol->transport, error))
    return FALSE;

  return TRUE;
}

gboolean consumer_service_client_ack (ConsumerServiceIf * iface, gboolean* _return, const AckResult * result, GError ** error)
{
  if (!consumer_service_client_send_ack (iface, result, error))
    return FALSE;
  if (!consumer_service_client_recv_ack (iface, _return, error))
    return FALSE;
  return TRUE;
}

static void
consumer_service_if_interface_init (ConsumerServiceIfInterface *iface)
{
  iface->pull = consumer_service_client_pull;
  iface->submit = consumer_service_client_submit;
  iface->get_consume_stats = consumer_service_client_get_consume_stats;
  iface->fetch = consumer_service_client_fetch;
  iface->ack = consumer_service_client_ack;
}

static void
consumer_service_client_init (ConsumerServiceClient *client)
{
  client->input_protocol = NULL;
  client->output_protocol = NULL;
}

static void
consumer_service_client_class_init (ConsumerServiceClientClass *cls)
{
  GObjectClass *gobject_class = G_OBJECT_CLASS (cls);
  GParamSpec *param_spec;

  gobject_class->set_property = consumer_service_client_set_property;
  gobject_class->get_property = consumer_service_client_get_property;

  param_spec = g_param_spec_object ("input_protocol",
                                    "input protocol (construct)",
                                    "Set the client input protocol",
                                    THRIFT_TYPE_PROTOCOL,
                                    G_PARAM_READWRITE);
  g_object_class_install_property (gobject_class,
                                   PROP_CONSUMER_SERVICE_CLIENT_INPUT_PROTOCOL, param_spec);

  param_spec = g_param_spec_object ("output_protocol",
                                    "output protocol (construct)",
                                    "Set the client output protocol",
                                    THRIFT_TYPE_PROTOCOL,
                                    G_PARAM_READWRITE);
  g_object_class_install_property (gobject_class,
                                   PROP_CONSUMER_SERVICE_CLIENT_OUTPUT_PROTOCOL, param_spec);
}

static void
consumer_service_handler_consumer_service_if_interface_init (ConsumerServiceIfInterface *iface);

G_DEFINE_TYPE_WITH_CODE (ConsumerServiceHandler, 
                         consumer_service_handler,
                         G_TYPE_OBJECT,
                         G_IMPLEMENT_INTERFACE (TYPE_CONSUMER_SERVICE_IF,
                                                consumer_service_handler_consumer_service_if_interface_init));

gboolean consumer_service_handler_pull (ConsumerServiceIf * iface, PullResponse ** _return, const PullRequest * request, PullException ** error, GError ** error)
{
  g_return_if_fail (IS_CONSUMER_SERVICE_HANDLER (iface));

  CONSUMER_SERVICE_HANDLER_GET_CLASS (iface)->pull (iface, _return, request, error, error);
}

gboolean consumer_service_handler_submit (ConsumerServiceIf * iface, gboolean* _return, const ConsumeResult * result, PullException ** error, GError ** error)
{
  g_return_if_fail (IS_CONSUMER_SERVICE_HANDLER (iface));

  CONSUMER_SERVICE_HANDLER_GET_CLASS (iface)->submit (iface, _return, result, error, error);
}

gboolean consumer_service_handler_get_consume_stats (ConsumerServiceIf * iface, GPtrArray ** _return, const ConsumeStatsRequest * request, PullException ** error, GError ** error)
{
  g_return_if_fail (IS_CONSUMER_SERVICE_HANDLER (iface));

  CONSUMER_SERVICE_HANDLER_GET_CLASS (iface)->get_consume_stats (iface, _return, request, error, error);
}

gboolean consumer_service_handler_fetch (ConsumerServiceIf * iface, FetchResponse ** _return, const FetchRequest * request, GError ** error)
{
  g_return_if_fail (IS_CONSUMER_SERVICE_HANDLER (iface));

  CONSUMER_SERVICE_HANDLER_GET_CLASS (iface)->fetch (iface, _return, request, error);
}

gboolean consumer_service_handler_ack (ConsumerServiceIf * iface, gboolean* _return, const AckResult * result, GError ** error)
{
  g_return_if_fail (IS_CONSUMER_SERVICE_HANDLER (iface));

  CONSUMER_SERVICE_HANDLER_GET_CLASS (iface)->ack (iface, _return, result, error);
}

static void
consumer_service_handler_consumer_service_if_interface_init (ConsumerServiceIfInterface *iface)
{
  iface->pull = consumer_service_handler_pull;
  iface->submit = consumer_service_handler_submit;
  iface->get_consume_stats = consumer_service_handler_get_consume_stats;
  iface->fetch = consumer_service_handler_fetch;
  iface->ack = consumer_service_handler_ack;
}

static void
consumer_service_handler_init (ConsumerServiceHandler *self)
{
  THRIFT_UNUSED_VAR (self);
}

static void
consumer_service_handler_class_init (ConsumerServiceHandlerClass *cls)
{
  cls->pull = NULL;
  cls->submit = NULL;
  cls->get_consume_stats = NULL;
  cls->fetch = NULL;
  cls->ack = NULL;
}

enum _ConsumerServiceProcessorProperties
{
  PROP_CONSUMER_SERVICE_PROCESSOR_0,
  PROP_CONSUMER_SERVICE_PROCESSOR_HANDLER
};

G_DEFINE_TYPE (ConsumerServiceProcessor,
               consumer_service_processor,
               THRIFT_TYPE_DISPATCH_PROCESSOR);

typedef gboolean (* ConsumerServiceProcessorProcessFunction) (ConsumerServiceProcessor *, 
                                                              gint32,
                                                              ThriftProtocol *,
                                                              ThriftProtocol *,
                                                              GError **);

static gboolean
consumer_service_processor_process_pull (ConsumerServiceProcessor *self,
                                         gint32 sequence_id,
                                         ThriftProtocol *input_protocol,
                                         ThriftProtocol *output_protocol,
                                         GError **error)
{
  gboolean result = TRUE;
  ThriftTransport * transport;
  ConsumerServicePullArgs * args =
    g_object_new (TYPE_CONSUMER_SERVICE_PULL_ARGS, NULL);

  g_object_get (input_protocol, "transport", &transport, NULL);

  if ((thrift_struct_read (THRIFT_STRUCT (args), input_protocol, error) != -1) &&
      (thrift_protocol_read_message_end (input_protocol, error) != -1) &&
      (thrift_transport_read_end (transport, error) != FALSE))
  {
    PullRequest * request;
    PullException * error = NULL;
    PullResponse * return_value;
    ConsumerServicePullResult * result_struct;

    g_object_get (args,
                  "request", &request,
                  NULL);

    g_object_unref (transport);
    g_object_get (output_protocol, "transport", &transport, NULL);

    result_struct = g_object_new (TYPE_CONSUMER_SERVICE_PULL_RESULT, NULL);
    g_object_get (result_struct, "success", &return_value, NULL);

    if (consumer_service_handler_pull (CONSUMER_SERVICE_IF (self->handler),
                                       &return_value,
                                       request,
                                       &error,
                                       error) == TRUE)
    {
      g_object_set (result_struct, "success", return_value, NULL);
      if (return_value != NULL)
        g_object_unref (return_value);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "pull",
                                               T_REPLY,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (result_struct),
                               output_protocol,
                               error) != -1));
    }
    else
    {
      if (error != NULL)
      {
        g_object_set (result_struct,
                      "error", error,
                      NULL);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "pull",
                                                 T_REPLY,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (result_struct),
                                 output_protocol,
                                 error) != -1));
      }
      else
      {
        if (*error == NULL)
          g_warning ("ConsumerService.pull implementation returned FALSE "
                     "but did not set an error");

        ThriftApplicationException *xception =
          g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
                        "type",    *error != NULL ? (*error)->code :
                                   THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN,
                        "message", *error != NULL ? (*error)->message : NULL,
                        NULL);
        g_clear_error (error);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "pull",
                                                 T_EXCEPTION,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (xception),
                                 output_protocol,
                                 error) != -1));

        g_object_unref (xception);
      }
    }

    if (request != NULL)
      g_object_unref (request);
    g_object_unref (result_struct);

    if (result == TRUE)
      result =
        ((thrift_protocol_write_message_end (output_protocol, error) != -1) &&
         (thrift_transport_write_end (transport, error) != FALSE) &&
         (thrift_transport_flush (transport, error) != FALSE));
  }
  else
    result = FALSE;

  g_object_unref (transport);
  g_object_unref (args);

  return result;
}

static gboolean
consumer_service_processor_process_submit (ConsumerServiceProcessor *self,
                                           gint32 sequence_id,
                                           ThriftProtocol *input_protocol,
                                           ThriftProtocol *output_protocol,
                                           GError **error)
{
  gboolean result = TRUE;
  ThriftTransport * transport;
  ConsumerServiceSubmitArgs * args =
    g_object_new (TYPE_CONSUMER_SERVICE_SUBMIT_ARGS, NULL);

  g_object_get (input_protocol, "transport", &transport, NULL);

  if ((thrift_struct_read (THRIFT_STRUCT (args), input_protocol, error) != -1) &&
      (thrift_protocol_read_message_end (input_protocol, error) != -1) &&
      (thrift_transport_read_end (transport, error) != FALSE))
  {
    ConsumeResult * result;
    PullException * error = NULL;
    gboolean return_value;
    ConsumerServiceSubmitResult * result_struct;

    g_object_get (args,
                  "result", &result,
                  NULL);

    g_object_unref (transport);
    g_object_get (output_protocol, "transport", &transport, NULL);

    result_struct = g_object_new (TYPE_CONSUMER_SERVICE_SUBMIT_RESULT, NULL);
    g_object_get (result_struct, "success", &return_value, NULL);

    if (consumer_service_handler_submit (CONSUMER_SERVICE_IF (self->handler),
                                         &return_value,
                                         result,
                                         &error,
                                         error) == TRUE)
    {
      g_object_set (result_struct, "success", return_value, NULL);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "submit",
                                               T_REPLY,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (result_struct),
                               output_protocol,
                               error) != -1));
    }
    else
    {
      if (error != NULL)
      {
        g_object_set (result_struct,
                      "error", error,
                      NULL);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "submit",
                                                 T_REPLY,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (result_struct),
                                 output_protocol,
                                 error) != -1));
      }
      else
      {
        if (*error == NULL)
          g_warning ("ConsumerService.submit implementation returned FALSE "
                     "but did not set an error");

        ThriftApplicationException *xception =
          g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
                        "type",    *error != NULL ? (*error)->code :
                                   THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN,
                        "message", *error != NULL ? (*error)->message : NULL,
                        NULL);
        g_clear_error (error);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "submit",
                                                 T_EXCEPTION,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (xception),
                                 output_protocol,
                                 error) != -1));

        g_object_unref (xception);
      }
    }

    if (result != NULL)
      g_object_unref (result);
    g_object_unref (result_struct);

    if (result == TRUE)
      result =
        ((thrift_protocol_write_message_end (output_protocol, error) != -1) &&
         (thrift_transport_write_end (transport, error) != FALSE) &&
         (thrift_transport_flush (transport, error) != FALSE));
  }
  else
    result = FALSE;

  g_object_unref (transport);
  g_object_unref (args);

  return result;
}

static gboolean
consumer_service_processor_process_get_consume_stats (ConsumerServiceProcessor *self,
                                                      gint32 sequence_id,
                                                      ThriftProtocol *input_protocol,
                                                      ThriftProtocol *output_protocol,
                                                      GError **error)
{
  gboolean result = TRUE;
  ThriftTransport * transport;
  ConsumerServiceGetConsumeStatsArgs * args =
    g_object_new (TYPE_CONSUMER_SERVICE_GET_CONSUME_STATS_ARGS, NULL);

  g_object_get (input_protocol, "transport", &transport, NULL);

  if ((thrift_struct_read (THRIFT_STRUCT (args), input_protocol, error) != -1) &&
      (thrift_protocol_read_message_end (input_protocol, error) != -1) &&
      (thrift_transport_read_end (transport, error) != FALSE))
  {
    ConsumeStatsRequest * request;
    PullException * error = NULL;
    GPtrArray * return_value;
    ConsumerServiceGetConsumeStatsResult * result_struct;

    g_object_get (args,
                  "request", &request,
                  NULL);

    g_object_unref (transport);
    g_object_get (output_protocol, "transport", &transport, NULL);

    result_struct = g_object_new (TYPE_CONSUMER_SERVICE_GET_CONSUME_STATS_RESULT, NULL);
    g_object_get (result_struct, "success", &return_value, NULL);

    if (consumer_service_handler_get_consume_stats (CONSUMER_SERVICE_IF (self->handler),
                                                    &return_value,
                                                    request,
                                                    &error,
                                                    error) == TRUE)
    {
      g_object_set (result_struct, "success", return_value, NULL);
      if (return_value != NULL)
        g_array_unref (return_value);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "getConsumeStats",
                                               T_REPLY,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (result_struct),
                               output_protocol,
                               error) != -1));
    }
    else
    {
      if (error != NULL)
      {
        g_object_set (result_struct,
                      "error", error,
                      NULL);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "getConsumeStats",
                                                 T_REPLY,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (result_struct),
                                 output_protocol,
                                 error) != -1));
      }
      else
      {
        if (*error == NULL)
          g_warning ("ConsumerService.getConsumeStats implementation returned FALSE "
                     "but did not set an error");

        ThriftApplicationException *xception =
          g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
                        "type",    *error != NULL ? (*error)->code :
                                   THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN,
                        "message", *error != NULL ? (*error)->message : NULL,
                        NULL);
        g_clear_error (error);

        result =
          ((thrift_protocol_write_message_begin (output_protocol,
                                                 "getConsumeStats",
                                                 T_EXCEPTION,
                                                 sequence_id,
                                                 error) != -1) &&
           (thrift_struct_write (THRIFT_STRUCT (xception),
                                 output_protocol,
                                 error) != -1));

        g_object_unref (xception);
      }
    }

    if (request != NULL)
      g_object_unref (request);
    g_object_unref (result_struct);

    if (result == TRUE)
      result =
        ((thrift_protocol_write_message_end (output_protocol, error) != -1) &&
         (thrift_transport_write_end (transport, error) != FALSE) &&
         (thrift_transport_flush (transport, error) != FALSE));
  }
  else
    result = FALSE;

  g_object_unref (transport);
  g_object_unref (args);

  return result;
}

static gboolean
consumer_service_processor_process_fetch (ConsumerServiceProcessor *self,
                                          gint32 sequence_id,
                                          ThriftProtocol *input_protocol,
                                          ThriftProtocol *output_protocol,
                                          GError **error)
{
  gboolean result = TRUE;
  ThriftTransport * transport;
  ConsumerServiceFetchArgs * args =
    g_object_new (TYPE_CONSUMER_SERVICE_FETCH_ARGS, NULL);

  g_object_get (input_protocol, "transport", &transport, NULL);

  if ((thrift_struct_read (THRIFT_STRUCT (args), input_protocol, error) != -1) &&
      (thrift_protocol_read_message_end (input_protocol, error) != -1) &&
      (thrift_transport_read_end (transport, error) != FALSE))
  {
    FetchRequest * request;
    FetchResponse * return_value;
    ConsumerServiceFetchResult * result_struct;

    g_object_get (args,
                  "request", &request,
                  NULL);

    g_object_unref (transport);
    g_object_get (output_protocol, "transport", &transport, NULL);

    result_struct = g_object_new (TYPE_CONSUMER_SERVICE_FETCH_RESULT, NULL);
    g_object_get (result_struct, "success", &return_value, NULL);

    if (consumer_service_handler_fetch (CONSUMER_SERVICE_IF (self->handler),
                                        &return_value,
                                        request,
                                        error) == TRUE)
    {
      g_object_set (result_struct, "success", return_value, NULL);
      if (return_value != NULL)
        g_object_unref (return_value);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "fetch",
                                               T_REPLY,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (result_struct),
                               output_protocol,
                               error) != -1));
    }
    else
    {
      if (*error == NULL)
        g_warning ("ConsumerService.fetch implementation returned FALSE "
                   "but did not set an error");

      ThriftApplicationException *xception =
        g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
                      "type",    *error != NULL ? (*error)->code :
                                 THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN,
                      "message", *error != NULL ? (*error)->message : NULL,
                      NULL);
      g_clear_error (error);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "fetch",
                                               T_EXCEPTION,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (xception),
                               output_protocol,
                               error) != -1));

      g_object_unref (xception);
    }

    if (request != NULL)
      g_object_unref (request);
    g_object_unref (result_struct);

    if (result == TRUE)
      result =
        ((thrift_protocol_write_message_end (output_protocol, error) != -1) &&
         (thrift_transport_write_end (transport, error) != FALSE) &&
         (thrift_transport_flush (transport, error) != FALSE));
  }
  else
    result = FALSE;

  g_object_unref (transport);
  g_object_unref (args);

  return result;
}

static gboolean
consumer_service_processor_process_ack (ConsumerServiceProcessor *self,
                                        gint32 sequence_id,
                                        ThriftProtocol *input_protocol,
                                        ThriftProtocol *output_protocol,
                                        GError **error)
{
  gboolean result = TRUE;
  ThriftTransport * transport;
  ConsumerServiceAckArgs * args =
    g_object_new (TYPE_CONSUMER_SERVICE_ACK_ARGS, NULL);

  g_object_get (input_protocol, "transport", &transport, NULL);

  if ((thrift_struct_read (THRIFT_STRUCT (args), input_protocol, error) != -1) &&
      (thrift_protocol_read_message_end (input_protocol, error) != -1) &&
      (thrift_transport_read_end (transport, error) != FALSE))
  {
    AckResult * result;
    gboolean return_value;
    ConsumerServiceAckResult * result_struct;

    g_object_get (args,
                  "result", &result,
                  NULL);

    g_object_unref (transport);
    g_object_get (output_protocol, "transport", &transport, NULL);

    result_struct = g_object_new (TYPE_CONSUMER_SERVICE_ACK_RESULT, NULL);
    g_object_get (result_struct, "success", &return_value, NULL);

    if (consumer_service_handler_ack (CONSUMER_SERVICE_IF (self->handler),
                                      &return_value,
                                      result,
                                      error) == TRUE)
    {
      g_object_set (result_struct, "success", return_value, NULL);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "ack",
                                               T_REPLY,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (result_struct),
                               output_protocol,
                               error) != -1));
    }
    else
    {
      if (*error == NULL)
        g_warning ("ConsumerService.ack implementation returned FALSE "
                   "but did not set an error");

      ThriftApplicationException *xception =
        g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
                      "type",    *error != NULL ? (*error)->code :
                                 THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN,
                      "message", *error != NULL ? (*error)->message : NULL,
                      NULL);
      g_clear_error (error);

      result =
        ((thrift_protocol_write_message_begin (output_protocol,
                                               "ack",
                                               T_EXCEPTION,
                                               sequence_id,
                                               error) != -1) &&
         (thrift_struct_write (THRIFT_STRUCT (xception),
                               output_protocol,
                               error) != -1));

      g_object_unref (xception);
    }

    if (result != NULL)
      g_object_unref (result);
    g_object_unref (result_struct);

    if (result == TRUE)
      result =
        ((thrift_protocol_write_message_end (output_protocol, error) != -1) &&
         (thrift_transport_write_end (transport, error) != FALSE) &&
         (thrift_transport_flush (transport, error) != FALSE));
  }
  else
    result = FALSE;

  g_object_unref (transport);
  g_object_unref (args);

  return result;
}

static gboolean
consumer_service_processor_dispatch_call (ThriftDispatchProcessor *dispatch_processor,
                                          ThriftProtocol *input_protocol,
                                          ThriftProtocol *output_protocol,
                                          gchar *method_name,
                                          gint32 sequence_id,
                                          GError **error)
{
  ConsumerServiceProcessorProcessFunction process_function; 
  gboolean dispatch_result = FALSE;

  ConsumerServiceProcessor *self = CONSUMER_SERVICE_PROCESSOR (dispatch_processor);
  ThriftDispatchProcessorClass *parent_class =
    g_type_class_peek_parent (CONSUMER_SERVICE_PROCESSOR_GET_CLASS (self));

  process_function = g_hash_table_lookup (self->process_map, method_name);
  if (process_function != NULL)
  {
    dispatch_result = (*process_function) (self,
                                           sequence_id,
                                           input_protocol,
                                           output_protocol,
                                           error);
  }
  else
  {
    dispatch_result = parent_class->dispatch_call (dispatch_processor,
                                                   input_protocol,
                                                   output_protocol,
                                                   method_name,
                                                   sequence_id,
                                                   error);
  }

  return dispatch_result;
}

static void
consumer_service_processor_set_property (GObject *object,
                                         guint property_id,
                                         const GValue *value,
                                         GParamSpec *pspec)
{
  ConsumerServiceProcessor *self = CONSUMER_SERVICE_PROCESSOR (object);

  switch (property_id)
  {
    case PROP_CONSUMER_SERVICE_PROCESSOR_HANDLER:
      if (self->handler != NULL)
        g_object_unref (self->handler);
      self->handler = g_value_get_object (value);
      g_object_ref (self->handler);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
  }
}

static void
consumer_service_processor_get_property (GObject *object,
                                         guint property_id,
                                         GValue *value,
                                         GParamSpec *pspec)
{
  ConsumerServiceProcessor *self = CONSUMER_SERVICE_PROCESSOR (object);

  switch (property_id)
  {
    case PROP_CONSUMER_SERVICE_PROCESSOR_HANDLER:
      g_value_set_object (value, self->handler);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
  }
}

static void
consumer_service_processor_dispose (GObject *gobject)
{
  ConsumerServiceProcessor *self = CONSUMER_SERVICE_PROCESSOR (gobject);

  if (self->handler != NULL)
  {
    g_object_unref (self->handler);
    self->handler = NULL;
  }

  G_OBJECT_CLASS (consumer_service_processor_parent_class)->dispose (gobject);
}

static void
consumer_service_processor_finalize (GObject *gobject)
{
  ConsumerServiceProcessor *self = CONSUMER_SERVICE_PROCESSOR (gobject);

  g_hash_table_destroy (self->process_map);

  G_OBJECT_CLASS (consumer_service_processor_parent_class)->finalize (gobject);
}

static void
consumer_service_processor_init (ConsumerServiceProcessor *self)
{
  self->handler = NULL;
  self->process_map = g_hash_table_new (g_str_hash, g_str_equal);

  g_hash_table_insert (self->process_map,
                       "pull", 
                       consumer_service_processor_process_pull);
  g_hash_table_insert (self->process_map,
                       "submit", 
                       consumer_service_processor_process_submit);
  g_hash_table_insert (self->process_map,
                       "getConsumeStats", 
                       consumer_service_processor_process_get_consume_stats);
  g_hash_table_insert (self->process_map,
                       "fetch", 
                       consumer_service_processor_process_fetch);
  g_hash_table_insert (self->process_map,
                       "ack", 
                       consumer_service_processor_process_ack);
}

static void
consumer_service_processor_class_init (ConsumerServiceProcessorClass *cls)
{
  GObjectClass *gobject_class = G_OBJECT_CLASS (cls);
  ThriftDispatchProcessorClass *dispatch_processor_class =
    THRIFT_DISPATCH_PROCESSOR_CLASS (cls);
  GParamSpec *param_spec;

  gobject_class->dispose = consumer_service_processor_dispose;
  gobject_class->finalize = consumer_service_processor_finalize;
  gobject_class->set_property = consumer_service_processor_set_property;
  gobject_class->get_property = consumer_service_processor_get_property;

  dispatch_processor_class->dispatch_call = consumer_service_processor_dispatch_call;
  cls->dispatch_call = consumer_service_processor_dispatch_call;

  param_spec = g_param_spec_object ("handler",
                                    "Service handler implementation",
                                    "The service handler implementation "
                                    "to which method calls are dispatched.",
                                    TYPE_CONSUMER_SERVICE_HANDLER,
                                    G_PARAM_READWRITE);
  g_object_class_install_property (gobject_class,
                                   PROP_CONSUMER_SERVICE_PROCESSOR_HANDLER,
                                   param_spec);
}
