lib/c_glib/src/thrift/c_glib/processor/thrift_multiplexed_processor.c - thrift - Git at Google

 /*
  * 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 <string.h>
 #include <stdio.h>

 #include <thrift/c_glib/thrift.h>
 #include <thrift/c_glib/processor/thrift_processor.h>
 #include <thrift/c_glib/processor/thrift_multiplexed_processor.h>
 #include <thrift/c_glib/protocol/thrift_multiplexed_protocol.h>
 #include <thrift/c_glib/protocol/thrift_stored_message_protocol.h>
 #include <thrift/c_glib/thrift_application_exception.h>

 G_DEFINE_TYPE(ThriftMultiplexedProcessor, thrift_multiplexed_processor, THRIFT_TYPE_PROCESSOR)


 enum
 {
   PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME = 1,
   PROP_THRIFT_MULTIPLEXED_PROCESSOR_END
 };

 static GParamSpec *thrift_multiplexed_processor_obj_properties[PROP_THRIFT_MULTIPLEXED_PROCESSOR_END] = { NULL, };


 static gboolean
 thrift_multiplexed_processor_register_processor_impl(ThriftProcessor *processor, const gchar * multiplexed_processor_name, ThriftProcessor * multiplexed_processor , GError **error)
 {
   ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(processor);
   g_hash_table_replace(self->multiplexed_services,
 		       g_strdup(multiplexed_processor_name),
 		       g_object_ref (multiplexed_processor));

   /* Make first registered become default */
   if(!self->default_processor_name){
       self->default_processor_name = g_strdup(multiplexed_processor_name);
   }
   return TRUE;
 }


 static gboolean
 thrift_multiplexed_processor_process_impl (ThriftProcessor *processor, ThriftProtocol *in,
 					   ThriftProtocol *out, GError **error)
 {
   gboolean retval = FALSE;
   gboolean token_error = FALSE;
   ThriftApplicationException *xception;
   ThriftStoredMessageProtocol *stored_message_protocol = NULL;
   ThriftMessageType message_type;
   ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(processor);
   ThriftProcessor *multiplexed_processor = NULL;
   ThriftTransport *transport;
   char *token=NULL;
   int token_index=0;
   char *state=NULL;
   gchar *fname=NULL;
   gint32 seqid, result;

   /* FIXME It seems that previous processor is not managing error correctly */
   if(*error!=NULL) {
       g_debug ("thrift_multiplexed_processor: last error not removed: %s",
       		   *error != NULL ? (*error)->message : "(null)");
       g_clear_error (error);
   }


   THRIFT_PROTOCOL_GET_CLASS(in)->read_message_begin(in, &fname, &message_type, &seqid, error);

   if(!(message_type == T_CALL || message_type == T_ONEWAY)) {
       g_set_error (error,
 		   THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
 		   THRIFT_MULTIPLEXED_PROCESSOR_ERROR_MESSAGE_TYPE,
 		   "message type invalid for this processor");
   }else{
       /* Split by the token */
       for (token = strtok_r(fname, THRIFT_MULTIPLEXED_PROTOCOL_DEFAULT_SEPARATOR, &state),
 	  token_index=0;
 	  token != NULL && !token_error;
 	  token = strtok_r(NULL, THRIFT_MULTIPLEXED_PROTOCOL_DEFAULT_SEPARATOR, &state),
 	      token_index++)
 	{
 	  switch(token_index){
 	    case 0:
 	      /* It should be the service name */
 	      multiplexed_processor = g_hash_table_lookup(self->multiplexed_services, token);
 	      if(multiplexed_processor==NULL){
 		  token_error=TRUE;
 	      }
 	      break;
 	    case 1:
 	      /* It should be the function name */
 	      stored_message_protocol = g_object_new (THRIFT_TYPE_STORED_MESSAGE_PROTOCOL,
 						      "protocol", in,
 						      "name", token,
 						      "type", message_type,
 						      "seqid", seqid,
 						      NULL);
 	      break;
 	    default:
 	      g_set_error (error,
 			   THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
 			   THRIFT_MULTIPLEXED_PROCESSOR_ERROR_MESSAGE_WRONGLY_MULTIPLEXED,
 			   "the message has more tokens than expected!");
 	      token_error=TRUE;
 	      break;
 	  }
 	}
       /* Set default */
       if(!stored_message_protocol &&
 	  !multiplexed_processor &&
 	  token_index==1 && self->default_processor_name){
 	  /* It should be the service name */
 	  multiplexed_processor = g_hash_table_lookup(self->multiplexed_services, self->default_processor_name);
 	  if(multiplexed_processor==NULL){
 	      g_set_error (error,
 			   THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
 			   THRIFT_MULTIPLEXED_PROCESSOR_ERROR_SERVICE_UNAVAILABLE,
 			   "service %s not available on this processor",
 			   self->default_processor_name);
 	  }else{
 	      /* Set the message name to the original name */
 	      stored_message_protocol = g_object_new (THRIFT_TYPE_STORED_MESSAGE_PROTOCOL,
 						      "protocol", in,
 						      "name", fname,
 						      "type", message_type,
 						      "seqid", seqid,
 						      NULL);
 	  }

       }

       if(stored_message_protocol!=NULL && multiplexed_processor!=NULL){
 	  retval = THRIFT_PROCESSOR_GET_CLASS (multiplexed_processor)->process (multiplexed_processor, (ThriftProtocol *) stored_message_protocol, out, error) ;
       }else{
 	  if(!error)
 	  g_set_error (error,
 		       THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
 		       THRIFT_MULTIPLEXED_PROCESSOR_ERROR_SERVICE_UNAVAILABLE,
 		       "service %s is not multiplexed in this processor",
 		       fname);
       }


   }

   if(!retval){
       /* By default, return an application exception to the client indicating the
           method name is not recognized. */
       /* Copied from dispach processor */

       if ((thrift_protocol_skip (in, T_STRUCT, error) < 0) ||
 	  (thrift_protocol_read_message_end (in, error) < 0))
 	return retval;

       g_object_get (in, "transport", &transport, NULL);
       result = thrift_transport_read_end (transport, error);
       g_object_unref (transport);
       if (result < 0) {
 	  /* We must free fname */
 	  g_free(fname);
 	  return retval;
       }

       if (thrift_protocol_write_message_begin (out,
 					       fname,
 					       T_EXCEPTION,
 					       seqid,
 					       error) < 0){
 	  /* We must free fname */
 	  g_free(fname);

 	  return retval;
       }


       xception =
 	  g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
 			"type",    THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN_METHOD,
 			"message", (*error)->message,
 			NULL);
       result = thrift_struct_write (THRIFT_STRUCT (xception),
 				    out,
 				    error);
       g_object_unref (xception);
       if ((result < 0) ||
 	  (thrift_protocol_write_message_end (out, error) < 0))
 	return retval;

       g_object_get (out, "transport", &transport, NULL);
       retval =
 	  ((thrift_transport_write_end (transport, error) >= 0) &&
 	      (thrift_transport_flush (transport, error) >= 0));
       g_object_unref (transport);
   }else{
       /* The protocol now has a copy we can free it */
       g_free(fname);

   }

   /*
   FIXME This makes everything fail, I don't know why.
   if(stored_message_protocol!=NULL){
 	  // After its use we must free it
 	  g_object_unref(stored_message_protocol);
   }
   */
   return retval;
 }

 /* define the GError domain for Thrift transports */
 GQuark
 thrift_multiplexed_processor_error_quark (void)
 {
   return g_quark_from_static_string (THRIFT_MULTIPLEXED_PROCESSOR_ERROR_DOMAIN);
 }


 static void
 thrift_multiplexed_processor_set_property (GObject      *object,
     guint         property_id,
     const GValue *value,
     GParamSpec   *pspec)
 {
   ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR (object);

   switch (property_id)
   {
   case PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME:
     self->default_processor_name = g_value_dup_string (value);
     break;

   default:
     /* We don't have any other property... */
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
     break;
   }
 }

 static void
 thrift_multiplexed_processor_get_property (GObject    *object,
     guint       property_id,
     GValue     *value,
     GParamSpec *pspec)
 {
   ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR (object);

   switch (property_id)
   {
   case PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME:
     g_value_set_string (value, self->default_processor_name);
     break;

   default:
     /* We don't have any other property... */
     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
     break;
   }
 }

 /* destructor */
 static void
 thrift_multiplexed_processor_finalize (GObject *object)
 {
   ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(object);

   /* Free our multiplexed hash table */
   g_hash_table_unref (self->multiplexed_services);
   self->multiplexed_services = NULL;

   if(self->default_processor_name){
       g_free(self->default_processor_name);
       self->default_processor_name=NULL;
   }

   /* Chain up to parent */
   if (G_OBJECT_CLASS (thrift_multiplexed_processor_parent_class)->finalize)
     (*G_OBJECT_CLASS (thrift_multiplexed_processor_parent_class)->finalize) (object);
 }

 /* class initializer for ThriftMultiplexedProcessor */
 static void
 thrift_multiplexed_processor_class_init (ThriftMultiplexedProcessorClass *cls)
 {
   /* Override */
   THRIFT_PROCESSOR_CLASS(cls)->process = thrift_multiplexed_processor_process_impl;
   GObjectClass *gobject_class = G_OBJECT_CLASS (cls);

   /* Object methods */
   gobject_class->set_property = thrift_multiplexed_processor_set_property;
   gobject_class->get_property = thrift_multiplexed_processor_get_property;
   gobject_class->finalize = thrift_multiplexed_processor_finalize;

   /* Class methods */
   cls->register_processor = thrift_multiplexed_processor_register_processor_impl;


   thrift_multiplexed_processor_obj_properties[PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME] =
       g_param_spec_string ("default",
           "Default service name the protocol points to where no multiplexed client used",
           "Set the default service name",
           NULL,
           (G_PARAM_READWRITE));

   g_object_class_install_properties (gobject_class,
 				     PROP_THRIFT_MULTIPLEXED_PROCESSOR_END,
       thrift_multiplexed_processor_obj_properties);

 }

 static void
 thrift_multiplexed_processor_init (ThriftMultiplexedProcessor *self)
 {

   /* Create our multiplexed services hash table */
   self->multiplexed_services = g_hash_table_new_full (
       g_str_hash,
       g_str_equal,
       g_free,
       g_object_unref);
   self->default_processor_name = NULL;
 }


 gboolean
 thrift_multiplexed_processor_register_processor(ThriftProcessor *processor, const gchar * multiplexed_processor_name, ThriftProcessor * multiplexed_processor , GError **error)
 {
   return THRIFT_MULTIPLEXED_PROCESSOR_GET_CLASS(processor)->register_processor(processor, multiplexed_processor_name, multiplexed_processor, error);
 }
	/*
	* 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 <string.h>
	#include <stdio.h>

	#include <thrift/c_glib/thrift.h>
	#include <thrift/c_glib/processor/thrift_processor.h>
	#include <thrift/c_glib/processor/thrift_multiplexed_processor.h>
	#include <thrift/c_glib/protocol/thrift_multiplexed_protocol.h>
	#include <thrift/c_glib/protocol/thrift_stored_message_protocol.h>
	#include <thrift/c_glib/thrift_application_exception.h>

	G_DEFINE_TYPE(ThriftMultiplexedProcessor, thrift_multiplexed_processor, THRIFT_TYPE_PROCESSOR)


	enum
	{
	PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME = 1,
	PROP_THRIFT_MULTIPLEXED_PROCESSOR_END
	};

	static GParamSpec *thrift_multiplexed_processor_obj_properties[PROP_THRIFT_MULTIPLEXED_PROCESSOR_END] = { NULL, };


	static gboolean
	thrift_multiplexed_processor_register_processor_impl(ThriftProcessor processor, const gchar multiplexed_processor_name, ThriftProcessor * multiplexed_processor , GError **error)
	{
	ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(processor);
	g_hash_table_replace(self->multiplexed_services,
	g_strdup(multiplexed_processor_name),
	g_object_ref (multiplexed_processor));

	/* Make first registered become default */
	if(!self->default_processor_name){
	self->default_processor_name = g_strdup(multiplexed_processor_name);
	}
	return TRUE;
	}


	static gboolean
	thrift_multiplexed_processor_process_impl (ThriftProcessor processor, ThriftProtocol in,
	ThriftProtocol out, GError *error)
	{
	gboolean retval = FALSE;
	gboolean token_error = FALSE;
	ThriftApplicationException *xception;
	ThriftStoredMessageProtocol *stored_message_protocol = NULL;
	ThriftMessageType message_type;
	ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(processor);
	ThriftProcessor *multiplexed_processor = NULL;
	ThriftTransport *transport;
	char *token=NULL;
	int token_index=0;
	char *state=NULL;
	gchar *fname=NULL;
	gint32 seqid, result;

	/* FIXME It seems that previous processor is not managing error correctly */
	if(*error!=NULL) {
	g_debug ("thrift_multiplexed_processor: last error not removed: %s",
	error != NULL ? (error)->message : "(null)");
	g_clear_error (error);
	}


	THRIFT_PROTOCOL_GET_CLASS(in)->read_message_begin(in, &fname, &message_type, &seqid, error);

	if(!(message_type == T_CALL \|\| message_type == T_ONEWAY)) {
	g_set_error (error,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR_MESSAGE_TYPE,
	"message type invalid for this processor");
	}else{
	/* Split by the token */
	for (token = strtok_r(fname, THRIFT_MULTIPLEXED_PROTOCOL_DEFAULT_SEPARATOR, &state),
	token_index=0;
	token != NULL && !token_error;
	token = strtok_r(NULL, THRIFT_MULTIPLEXED_PROTOCOL_DEFAULT_SEPARATOR, &state),
	token_index++)
	{
	switch(token_index){
	case 0:
	/* It should be the service name */
	multiplexed_processor = g_hash_table_lookup(self->multiplexed_services, token);
	if(multiplexed_processor==NULL){
	token_error=TRUE;
	}
	break;
	case 1:
	/* It should be the function name */
	stored_message_protocol = g_object_new (THRIFT_TYPE_STORED_MESSAGE_PROTOCOL,
	"protocol", in,
	"name", token,
	"type", message_type,
	"seqid", seqid,
	NULL);
	break;
	default:
	g_set_error (error,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR_MESSAGE_WRONGLY_MULTIPLEXED,
	"the message has more tokens than expected!");
	token_error=TRUE;
	break;
	}
	}
	/* Set default */
	if(!stored_message_protocol &&
	!multiplexed_processor &&
	token_index==1 && self->default_processor_name){
	/* It should be the service name */
	multiplexed_processor = g_hash_table_lookup(self->multiplexed_services, self->default_processor_name);
	if(multiplexed_processor==NULL){
	g_set_error (error,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR_SERVICE_UNAVAILABLE,
	"service %s not available on this processor",
	self->default_processor_name);
	}else{
	/* Set the message name to the original name */
	stored_message_protocol = g_object_new (THRIFT_TYPE_STORED_MESSAGE_PROTOCOL,
	"protocol", in,
	"name", fname,
	"type", message_type,
	"seqid", seqid,
	NULL);
	}

	}

	if(stored_message_protocol!=NULL && multiplexed_processor!=NULL){
	retval = THRIFT_PROCESSOR_GET_CLASS (multiplexed_processor)->process (multiplexed_processor, (ThriftProtocol *) stored_message_protocol, out, error) ;
	}else{
	if(!error)
	g_set_error (error,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR,
	THRIFT_MULTIPLEXED_PROCESSOR_ERROR_SERVICE_UNAVAILABLE,
	"service %s is not multiplexed in this processor",
	fname);
	}


	}

	if(!retval){
	/* By default, return an application exception to the client indicating the
	method name is not recognized. */
	/* Copied from dispach processor */

	if ((thrift_protocol_skip (in, T_STRUCT, error) < 0) \|\|
	(thrift_protocol_read_message_end (in, error) < 0))
	return retval;

	g_object_get (in, "transport", &transport, NULL);
	result = thrift_transport_read_end (transport, error);
	g_object_unref (transport);
	if (result < 0) {
	/* We must free fname */
	g_free(fname);
	return retval;
	}

	if (thrift_protocol_write_message_begin (out,
	fname,
	T_EXCEPTION,
	seqid,
	error) < 0){
	/* We must free fname */
	g_free(fname);

	return retval;
	}


	xception =
	g_object_new (THRIFT_TYPE_APPLICATION_EXCEPTION,
	"type", THRIFT_APPLICATION_EXCEPTION_ERROR_UNKNOWN_METHOD,
	"message", (*error)->message,
	NULL);
	result = thrift_struct_write (THRIFT_STRUCT (xception),
	out,
	error);
	g_object_unref (xception);
	if ((result < 0) \|\|
	(thrift_protocol_write_message_end (out, error) < 0))
	return retval;

	g_object_get (out, "transport", &transport, NULL);
	retval =
	((thrift_transport_write_end (transport, error) >= 0) &&
	(thrift_transport_flush (transport, error) >= 0));
	g_object_unref (transport);
	}else{
	/* The protocol now has a copy we can free it */
	g_free(fname);

	}

	/*
	FIXME This makes everything fail, I don't know why.
	if(stored_message_protocol!=NULL){
	// After its use we must free it
	g_object_unref(stored_message_protocol);
	}
	*/
	return retval;
	}

	/* define the GError domain for Thrift transports */
	GQuark
	thrift_multiplexed_processor_error_quark (void)
	{
	return g_quark_from_static_string (THRIFT_MULTIPLEXED_PROCESSOR_ERROR_DOMAIN);
	}


	static void
	thrift_multiplexed_processor_set_property (GObject *object,
	guint property_id,
	const GValue *value,
	GParamSpec *pspec)
	{
	ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR (object);

	switch (property_id)
	{
	case PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME:
	self->default_processor_name = g_value_dup_string (value);
	break;

	default:
	/* We don't have any other property... */
	G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
	break;
	}
	}

	static void
	thrift_multiplexed_processor_get_property (GObject *object,
	guint property_id,
	GValue *value,
	GParamSpec *pspec)
	{
	ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR (object);

	switch (property_id)
	{
	case PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME:
	g_value_set_string (value, self->default_processor_name);
	break;

	default:
	/* We don't have any other property... */
	G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
	break;
	}
	}

	/* destructor */
	static void
	thrift_multiplexed_processor_finalize (GObject *object)
	{
	ThriftMultiplexedProcessor *self = THRIFT_MULTIPLEXED_PROCESSOR(object);

	/* Free our multiplexed hash table */
	g_hash_table_unref (self->multiplexed_services);
	self->multiplexed_services = NULL;

	if(self->default_processor_name){
	g_free(self->default_processor_name);
	self->default_processor_name=NULL;
	}

	/* Chain up to parent */
	if (G_OBJECT_CLASS (thrift_multiplexed_processor_parent_class)->finalize)
	(*G_OBJECT_CLASS (thrift_multiplexed_processor_parent_class)->finalize) (object);
	}

	/* class initializer for ThriftMultiplexedProcessor */
	static void
	thrift_multiplexed_processor_class_init (ThriftMultiplexedProcessorClass *cls)
	{
	/* Override */
	THRIFT_PROCESSOR_CLASS(cls)->process = thrift_multiplexed_processor_process_impl;
	GObjectClass *gobject_class = G_OBJECT_CLASS (cls);

	/* Object methods */
	gobject_class->set_property = thrift_multiplexed_processor_set_property;
	gobject_class->get_property = thrift_multiplexed_processor_get_property;
	gobject_class->finalize = thrift_multiplexed_processor_finalize;

	/* Class methods */
	cls->register_processor = thrift_multiplexed_processor_register_processor_impl;


	thrift_multiplexed_processor_obj_properties[PROP_THRIFT_MULTIPLEXED_PROCESSOR_DEFAULT_SERVICE_NAME] =
	g_param_spec_string ("default",
	"Default service name the protocol points to where no multiplexed client used",
	"Set the default service name",
	NULL,
	(G_PARAM_READWRITE));

	g_object_class_install_properties (gobject_class,
	PROP_THRIFT_MULTIPLEXED_PROCESSOR_END,
	thrift_multiplexed_processor_obj_properties);

	}

	static void
	thrift_multiplexed_processor_init (ThriftMultiplexedProcessor *self)
	{

	/* Create our multiplexed services hash table */
	self->multiplexed_services = g_hash_table_new_full (
	g_str_hash,
	g_str_equal,
	g_free,
	g_object_unref);
	self->default_processor_name = NULL;
	}


	gboolean
	thrift_multiplexed_processor_register_processor(ThriftProcessor processor, const gchar multiplexed_processor_name, ThriftProcessor * multiplexed_processor , GError **error)
	{
	return THRIFT_MULTIPLEXED_PROCESSOR_GET_CLASS(processor)->register_processor(processor, multiplexed_processor_name, multiplexed_processor, error);
	}