You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@etch.apache.org by fi...@apache.org on 2010/07/21 18:58:33 UTC
svn commit: r966314 [19/41] - in /incubator/etch/branches/etch-c: ./
binding-c/ binding-c/compiler/ binding-c/compiler/src/
binding-c/compiler/src/main/ binding-c/compiler/src/main/java/
binding-c/compiler/src/main/java/org/ binding-c/compiler/src/main...
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_stub.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_stub.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_stub.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_stub.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,877 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_stub.c
+ * contains the base runtime code for either type of stub.
+ *
+ * the generated stub_xxxx_server will define the thread procedures for each message
+ * type in the service. for each such type, it will set the virtual stub helpers in
+ * the type objects, to point to these thread functions. each such thread procedure
+ * will call the user-coded calculation in the impl_xxxx_server for that message
+ * type; for example, for an "add" message, it will get the fields from the service
+ * value factory which are the arguments to the calculation, call the calculation in
+ * the server impl, insert the result of the calculation into the reply message, and
+ * call the delivery service transport_message to send the reply back to the sender.
+ */
+
+#include "etch.h"
+#include "etch_svcobj_masks.h"
+#include "etch_stub.h"
+#include "etch_exception.h"
+#include "etch_objecttypes.h"
+#include "etch_log.h"
+#include "etch_mem.h"
+#include "etch_object.h"
+#include "etch_message.h"
+static const char* LOG_CATEGORY = "etch_stub";
+
+int etchstub_session_notify_ex (void*, void*, etch_event*);
+int etchstub_session_message(void*, etch_who*, etch_message*);
+int etchstub_session_control(void*, etch_event*, etch_object*);
+int etchstub_session_notify (void*, etch_event*);
+etch_object* etchstub_session_query (void*, etch_query*);
+i_objsession* etchstub_get_session_callbacks_from (void* _obj);
+
+char* scstr = "session control", *snstr = "session notify", *sqstr = "session query";
+char* logmask1 = "%s not routed to client\n", *logmask2 = "%s not registered\n";
+
+
+
+#if(0)
+____________________________________________________________________________________
+the inheritance hierarchy becomes indefinite at this point. a stub should be able
+to query a service object for a particular interface, down through its inheritance
+chain. for example, we need to find who has implemented obj_session and call methods
+on it. a use case is a custom transport implemented as a dynamically loaded dll.
+the stub needs therefore to be able to call into objects it has never seen before.
+
+for the present we will make some assumptions as to the inheritance hierarchy, that
+being that given a service interface, we know its inheritance hierarchy. following
+is a diagram illustrating the inheritance of a service interface named XXXX.
+all arrows direct down unless indicated by an arrowhead. (e) indicates extends,
+(i) indicates implements. this diagram is upside-down to the way we normally see
+the hierarchy.
+ (obj)
+ |
+ XXXX <- - - - - - - - - - (stub)
+ | |
+ (e) --------- ---------- (e)
+ | |
+ XXXX_server remote_XXXX
+ | \ (i) | (e)
+ (i) | ------> remote_XXXX_server
+ obj_session <-- base_XXXX_server
+ (e) |
+ impl_XXXX_server
+____________________________________________________________________________________
+#endif
+
+
+#if(0)
+
+IMPL_XXXX_SERVER
+ - ctor(REMOTE_XXXX_CLIENT)
+ - usermethod_1_impl(); ...
+ - usermethod_n_impl();
+
+ BASE_XXXX_SERVER
+ - usermethod_1_stub(); ...
+ - usermethod_n_stub();
+ OBJSESSION
+ - _session_control; _session_notify; _session_query; (stubs)
+ XXXX_SERVER
+ - any server-directed methods go here
+ XXXX
+ - see user method impls above
+
+
+STUBXXXX<T extends XXXX> extends STUBBASE<T>
+ - ctor(DeliveryService, pool, pool);
+ - stubhelper();
+
+ STUBBASE<T> (DeliveryService, pool, pool)
+ T obj;
+ DeliveryService, Pool, Pool;
+ stub_helper();
+
+ STUBPOOLRUNNABLE
+ - set(stub_helper, who, msg); (set state, not in java)
+ - stubhelper(stub, delsvc, obj, who, msg);
+
+ SESSIONMESSAGE
+ - session_message(who, message);
+ { stubhelper = message.type.get_helper()
+ stubhelper(delsvc, obj, who, msg); // the stubhelper impl calls threadpool run()
+ }
+ - session_control; session_notify; session_query;
+
+
+STUBXXXSERVER (generated, hand-codable)
+ - ctor(DeliveryService, XXXXSERVER, pool, pool);
+ - for each type in the vf,
+ -- implement a run method (threadproc) for the message type
+ -- implement a stub helper method which calls the threadpool run using the threadproc above
+
+ STUBXXXX
+
+#endif
+
+
+/* - - - - - - - - - - - - - - - - - -
+ * stub base method implementations
+ * - - - - - - - - - - - - - - - - - -
+ */
+
+/**
+ * destroy_stub()
+ * etch_stub destructor. destroy stub base object.
+ * the stub base does not destroy its xxxx_either_stub wrapper.
+ */
+int destroy_stub(void* data)
+{
+ etch_stub* thisx = (etch_stub*)data;
+
+ if (!is_etchobj_static_content(thisx))
+ {
+ /* note that a delivery service session is the stub's session.
+ * the delivery service owns the stub, and the stub owns this shared
+ * session. the stub wrapper is the session interface's thisx.
+ * the delivery service gets xxxx_either_stub* stubbobj from its
+ * isessionmsg->thisx, and etch_stub* stubbase from stubobj.stub_base.
+ * destroying the stub will destroy the shared i_sessionmessage, so
+ * the delivery service's session will become invalid once its stub
+ * is destroyed. see transport.destroy_delivery_service_stub().
+ */
+ etch_object_destroy(thisx->isessionmsg);
+
+ if (thisx->obj && thisx->is_implobj_owned)
+ etch_object_destroy(thisx->obj);
+ }
+
+ return destroy_objectex((etch_object*) thisx);
+}
+
+
+/**
+ * new_stub()
+ * etch_stub (stub base) constructor.
+ */
+etch_stub* new_stub (void* implobj, unsigned char stubtype,
+ i_delivery_service* ids, etch_threadpool* qp, etch_threadpool* fp)
+{
+ i_sessionmessage* ism = NULL;
+
+ etch_stub* stubbase = (etch_stub*) new_object
+ (sizeof(etch_stub), ETCHTYPEB_STUB, CLASSID_STUB);
+
+ ETCH_ASSERT(implobj && stubtype && qp && fp);
+ ETCH_ASSERT(is_etch_ideliverysvc(ids));
+
+ stubbase->stub_type = stubtype;
+ ((etch_object*)stubbase)->destroy = destroy_stub;
+
+ stubbase->obj = implobj; /* server or client impl e.g. i_xxxx_server* */
+ stubbase->delivery_service = ids;
+
+ /* instantiate i_sessionmessage session interface. note that the interface's
+ * thisx pointer is this base stub object, and that the base stub's container
+ * xxxx_either_stub* is the base stub.stubobj.
+ */
+ ism = new_sessionmsg_interface (stubbase, etchstub_session_message, NULL);
+ stubbase->isessionmsg = ism;
+ stubbase->session_message = ism->session_message = etchstub_session_message;
+ stubbase->session_control = ism->session_control = etchstub_session_control;
+ stubbase->session_notify = ism->session_notify = etchstub_session_notify;
+ stubbase->session_query = ism->session_query = etchstub_session_query;
+
+ /* set delivery service session interface to be the stub's i_sessionmessage.
+ * in a c binding translation from java binding, a reference to a stub object
+ * is only via via the delivery service's session_message interface, shared
+ * between the stub and the delivery service.
+ * the delivery service must have provided i_transportmessage implementations,
+ * in particular for set_session(). the instantiator of the delivery service
+ * must therefore override the delivery service's i_transportmessage.
+
+ * fyi: ids->itm->thisx is mailbox manager.
+ * ids->itm is mboxmgr's transport interface, which is the delivery service.
+ * set session of next lower level (delivery service) to the stub's session.
+ */
+ ids->itm->set_session (ids, stubbase->isessionmsg);
+
+ /* copy impl's i_objsession to the stub for convenience */
+ stubbase->impl_callbacks = etchstub_get_session_callbacks_from (implobj);
+
+ return stubbase;
+}
+
+
+/**
+ * etchstub_get_session_callbacks_from()
+ * extract objsession interface from specified stub implementor object.
+ */
+i_objsession* etchstub_get_session_callbacks_from (void* _obj)
+{
+ i_objsession* iobjsession = NULL;
+ etch_object* obj = (etch_object*) _obj;
+ const int this_objtype = obj? ((etch_object*)obj)->obj_type: 0;
+
+ switch(this_objtype)
+ {
+ case ETCHTYPEB_EXESERVERIMPL:
+ { xxxx_server_impl* server_impl = (xxxx_server_impl*) obj;
+ iobjsession = server_impl->iobjsession;
+ break;
+ }
+
+ case ETCHTYPEB_EXECLIENTIMPL:
+ { xxxx_client_impl* client_impl = (xxxx_client_impl*) obj;
+ iobjsession = client_impl->iobjsession;
+ break;
+ }
+
+ case ETCHTYPEB_EXESERVERBASE:
+ { i_xxxx_server* iserver = (i_xxxx_server*) obj;
+ xxxx_server_impl* server_impl = (xxxx_server_impl*) iserver->thisx;
+ ETCH_ASSERT(server_impl && ((etch_object*)server_impl)->obj_type == ETCHTYPEB_EXESERVERIMPL);
+ iobjsession = server_impl->iobjsession;
+ break;
+ }
+
+ case ETCHTYPEB_EXECLIENTBASE:
+ { i_xxxx_client* iclient = (i_xxxx_client*) obj;
+ xxxx_client_impl* client_impl = (xxxx_client_impl*) iclient->thisx;
+ ETCH_ASSERT(client_impl && ((etch_object*)client_impl)->obj_type == ETCHTYPEB_EXECLIENTIMPL);
+ iobjsession = client_impl->iobjsession;
+ break;
+ }
+ }
+
+ return iobjsession;
+}
+
+/**
+ * etchstub_put_resultobj
+ * insert specified result object to message.
+ * @param replymsg the message, caller retains.
+ * @param resultobj the result object, caller relinquishes regardless of result.
+ * @return 0 success, -1 failure.
+ */
+int etchstub_put_resultobj (etch_stub* stub, etch_message* replymsg, etch_object* resultobj)
+{
+ int result = 0;
+ etch_field* key_result = builtins._mf_result;
+
+ /* check if the message already has a result object, i.e., an exception */
+ if (message_get (replymsg, key_result))
+ etch_object_destroy(resultobj); /* if so, discard the passed result */
+ else /* resultobj is relinquished here regardless of result */
+ result = message_put (replymsg, clone_field(key_result), (etch_object*) resultobj);
+
+ return result;
+}
+
+
+/**
+ * etchstub_send_reply
+ * private method invoked by all "stub helpers" (message logic implementation
+ * runners) for messages requiring a reply, in order to instantiate a reply
+ * message and transmit it back to sender.
+ */
+int etchstub_send_reply (etch_stub* stub, i_delivery_service* dsvc, etch_who* whofrom, etch_message* msg, etch_object* resultobj, boolean putResultObj)
+{
+ int result = 0;
+ const int is_exception_resobj = is_etch_exception(resultobj);
+
+ etch_type* newtype = NULL; /* see comment following */
+
+ /* instantiate the reply message
+ * note that if we are called for a one-way message, it is only if an
+ * exception was thrown by the implementation, and the exception is
+ * therefore to be the reply. in such a case, the etch compiler generates
+ * a message type as newtype above, which is the second parameter to
+ * message_reply(), in order that in_reply_to can be instantiated.
+ * (our example below makes newtype the 1-way exception reply type in
+ * this case). for two-way messages, newtype, and thus the second parameter
+ * to message_reply() is null.
+ */
+ etch_message* replymsg = message_reply (msg, newtype);
+
+ if (NULL == replymsg && is_exception_resobj)
+ { newtype = builtins._mt__exception;
+ replymsg = message_reply (msg, newtype);
+ }
+
+ if (NULL == replymsg) {
+ etch_type* replytype = newtype? newtype: message_type(msg);
+ char* logmask = "could not create reply message for type %s\n";
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, logmask, replytype->aname);
+ return -1;
+ }
+
+ if(putResultObj) {
+ result = etchstub_put_resultobj (stub, replymsg, resultobj);
+ if(result){
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "invalid return value.\n");
+ return result;
+ }
+ }
+
+ if (0 == result) { /* forward reply message for serialization */
+ result = dsvc->itm->transport_message (dsvc, whofrom, replymsg);
+ if(result){
+ etch_object_destroy(replymsg);
+ }
+ }
+
+ return result;
+}
+
+
+/**
+ * etchstub_validate_args
+ * private method invoked by all "stub helpers" to validate parameters
+ * to the individual stub helper and populate some objects for the helper.
+ */
+int etchstub_validate_args (etch_stub* stub, i_delivery_service* dsvc,
+ etch_message* msg, void* client, default_value_factory** vf,
+ void** vfimpl, void** climpl)
+{
+ int result = 0;
+ i_xxxx_client* iclient = NULL;
+ etch_object* clientimpl = NULL, *valufactimpl = NULL;
+ /* any assertion failure in this method indicates a coding error in etch core */
+ ETCH_ASSERT(stub && dsvc && client && msg && stub->params && vf && vfimpl && climpl);
+
+ iclient = (i_xxxx_client*) client;
+ clientimpl = iclient->thisx;
+ //HCG Check this for client!
+ //ETCH_ASSERT(is_etch_server_impl(clientimpl));
+ *climpl = clientimpl;
+
+ *vf = (default_value_factory*) stub->params->in_valufact;
+ ETCH_ASSERT(is_etch_valuefact(*vf));
+ valufactimpl = (*vf)->impl;
+ ETCH_ASSERT(is_etch_valuefactimpl(valufactimpl));
+ *vfimpl = valufactimpl;
+
+ return result;
+}
+
+
+/**
+ * etchstub_session_notify_ex()
+ *
+ * @param thisx the object which will receive an exception if any.
+ * may be the same object as _obj. caller retains.
+ *
+ * @param _obj caller retains. usually a xxxx_server_impl. may be null, or
+ * may be the same object as thisx. obj will belong to a known and limited set
+ * of classes, e.g. ETCHTYPEB_EXESERVERIMPL, so we can test _obj.obj_type
+ * if needed, and cast _obj to one of the mask objects (etch_svcobj_masks.h)
+ * such as xxxx_server_impl, in order to reference the maskable content.
+ *
+ * @param evt a notification event or a throwable exception, caller relinquishes.
+ *
+ * @return 0 success, -1 failure.
+ */
+int etchstub_session_notify_ex (void* thisx, void* _obj, etch_event* evt)
+{
+ int result = -1, is_evtparm_relinquished = FALSE;
+
+ /* this call arrives from type "stub helpers" implemented in xxxx_server_stub.
+ * these are thread procedures with arguments delivery service, some object
+ * _obj (possibly implementing objsession or throwable), who, and message,
+ * viz: int (*stubhelper) (stub, deliverysvc, _obj, sender, message);
+ * so from argument _obj, we extract the objsession interface and if present,
+ * call its _session_notify.
+ */
+
+ /* a server implementation (xxxx_server_impl) can request notifications of
+ * exceptions and the like by implementing and registering i_objsession
+ * callbacks. the presence of these function pointers in the implementation
+ * serves as the indicator of whether to forward the notification.
+ */
+ i_objsession* impl_callbacks = etchstub_get_session_callbacks_from (_obj);
+
+ etch_session_notify impl_session_notify = impl_callbacks?
+ impl_callbacks->_session_notify: NULL;
+
+ if (impl_session_notify) /* if impl has requested this event, forward it */
+ { /* event is relinquished to notify handlers by contract */
+ result = impl_session_notify (impl_callbacks, evt);
+ is_evtparm_relinquished = TRUE;
+ }else {
+ if(((xxxx_either_impl*)_obj)->_session_notify) {
+ result = ((xxxx_either_impl*)_obj)->_session_notify ((xxxx_either_impl*)_obj, evt);
+ is_evtparm_relinquished = TRUE;
+ }
+ }
+
+ if (0 != result)
+ {
+ char* logmask = impl_session_notify? logmask1: logmask2;
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, logmask, snstr);
+ }
+
+ if (!is_evtparm_relinquished)
+ {
+ etch_object_destroy(evt);
+ evt = NULL;
+ //ETCHOBJ_DESTROY(evt);
+ }
+
+ return result;
+}
+
+
+/* - - - - - - - - - - - - - - - - - - - - -
+ * stub helper threadpool execution support
+ * - - - - - - - - - - - - - - - - - - - - -
+ */
+
+/**
+ * etchstub_loghelper()
+ * convenience method to log entry or exit of a stub helper function.
+ */
+void etchstub_loghelper(opaque_stubhelper f, const int result, const int thread_id, const int is_start)
+{
+ if(is_start) {
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "start stub runner %x [%d]\n", f, thread_id);
+ } else {
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "exit stub runner %x (%d) [%d]\n", f, result, thread_id);
+ }
+}
+
+/**
+ * etchstub_proxy_threadproc()
+ * a thread procedure conforming to the signature expected by etch_threadpool,
+ * which accepts a stub parameter bundle as params.data, unbundles the parameters,
+ * and invokes the included stub helper function with those parameters.
+ */
+void etchstub_proxy_threadproc (void* data)
+{
+ etch_thread_params* tp = (etch_thread_params*)data;
+ int result = 0;
+ etchstub_runparams* p = tp? tp->data: NULL;
+ ETCH_ASSERT(p && p->run && (p->sig == ETCH_STUBPARAMS_SIGNATURE));
+ etchstub_loghelper (p->run, 0, tp->etch_thread_id, 1);
+
+ /* run "stub helper" procedure */
+ result = p->run (p->stub, p->ds, p->server, p->who, p->msg);
+
+ etchstub_loghelper (p->run, result, tp->etch_thread_id, 0);
+
+ //tp->etchobj->destroy(tp->etchobj);
+ etch_free(p);
+
+
+}
+
+
+/**
+ * new_etch_stubparams()
+ * bundle up stub runner parameters in order they can become
+ * etch_thread_params.data, passed in to the etchstub_proxy_threadproc.
+ */
+etchstub_runparams* new_etch_stubparams (etch_stub* stub, opaque_stubhelper runproc,
+ i_delivery_service* ds, i_xxxx_server* server, etch_who* whofrom, etch_message* msg)
+{
+ etchstub_runparams* p = etch_malloc(sizeof(etchstub_runparams), 0);
+ p->sig = ETCH_STUBPARAMS_SIGNATURE;
+ p->stub = stub;
+ p->run = runproc;
+ p->ds = ds;
+ p->who = whofrom;
+ p->msg = msg;
+ p->server = server;
+ return p;
+}
+
+
+/**
+ * etchstub_run()
+ * execute the service method helper function (stub helper). if no threadpool
+ * is specified, execute the helper function inline (on caller's thread);
+ * otherwise wrap the stub helper arguments and launch a thread on the supplied
+ * threadpool to execute the helper function.
+ * @param stub caller retains.
+ * @param runproc the service function for this message type.
+ * @param threadpool caller retains.
+ * @param server caller retains.
+ * @param who caller retains.
+ * @param msg caller retains.
+ * @return 0 success, -1 failure.
+ */
+int etchstub_run (etch_stub* stub, opaque_stubhelper runproc, etch_threadpool* threadpool,
+ i_xxxx_server* server, etch_who* who, etch_message* msg)
+{
+ int result = 0;
+ i_delivery_service* ds = stub->delivery_service;
+ ETCH_ASSERT(runproc && ds && server && msg);
+
+ if (threadpool)
+ {
+ /* a pointer to this parameter bundle will become etch_thread_params.data.
+ * etch_thread_params.is_own_data determines if the thread frees data on exit.
+ * is_own_data assumes the value of threadpool.is_free_data, which is true
+ * by default. so unless we have reset is_free_data on this pool, the pool
+ * thread will free this allocation at thread exit. and since also by default
+ * pool->is_data_etchobject is false, the deallocation will use etch_free;
+ */
+ etchstub_runparams* p = new_etch_stubparams(stub, runproc, ds, server, who, msg);
+ etch_thread* poolthread = NULL;
+ threadpool->is_manual_start = TRUE;
+
+ /* run stub helper function on a pool thread */
+ poolthread = threadpool->run (threadpool, etchstub_proxy_threadproc, p);
+
+ //poolthread->is_joined = TRUE;
+ //poolthread->on_exit()
+ poolthread->start(poolthread);
+// threadpool->is_manual_start = FALSE;
+
+ //if (poolthread) /* switch context and block on pool thread exit */
+ //{ const int id = poolthread->params.etch_thread_id;
+ // //etch_log(etchstub, etch_log_xdebug, "stub runner joining pool thread %d ...\n", id);
+ // etch_join (poolthread);
+ //}
+ //else result = -1;
+ }
+ else
+ { etchstub_loghelper(runproc, 0, 0, 1);
+
+ /* run stub helper function on this thread */
+ result = runproc (stub, ds, server, who, msg);
+
+ etchstub_loghelper(runproc, result, 0, 0);
+
+ }
+ return result;
+}
+
+
+/* - - - - - - - - - -
+ * i_sessionmessage
+ * - - - - - - - - - -
+ */
+
+/**
+ * etchstub_session_message()
+ * @param whofrom caller retains, can be null.
+ * @param msg caller relinquishes
+ * @return 0 (message handled), or -1 (error, closed, or timeout)
+ */
+int etchstub_session_message (void* data, etch_who* whofrom, etch_message* msg)
+{
+ etch_stub* thisx = (etch_stub*)data;
+ xxxx_either_stub* stubimpl = (xxxx_either_stub*) thisx->stubobj;
+ i_xxxx_server* serverimpl = NULL;
+ etch_session* sessionparams = NULL;
+ opaque_stubhelper stubhelper = NULL;
+ etch_threadpool* threadpool = NULL;
+ int result = -1, session_id = 0;
+ unsigned char async_mode = 0;
+
+ /* get the message type object associated with this message */
+ etch_type* thistype = message_type(msg);
+ ETCH_ASSERT(thistype);
+ ETCH_ASSERT(is_etch_stub(stubimpl));
+
+ session_id = stubimpl->owner_id;
+ do
+ {
+ /* get the message type's stub helper (runner) function.
+ * the stub helper is a pointer to a thread procedure function specific
+ * to the message type, which executes the associated service method.
+ */
+ if (NULL == (stubhelper = etchtype_get_type_stubhelper (thistype)))
+ { char* msgmask = "type '%s' missing stub runner procedure\n";
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, msgmask, thistype->aname);
+ break;
+ }
+
+ /* get the thread pool, if any, appropriate for the
+ * configured mode of execution for this message type
+ */
+ switch(async_mode = etchtype_get_async_mode (thistype))
+ {
+ //default: threadpool = thisx->params->qpool; break;
+ case ETCH_ASYNCMODE_QUEUED: threadpool = thisx->params->qpool; break;
+ case ETCH_ASYNCMODE_FREE: threadpool = thisx->params->fpool; break;
+ default: threadpool = NULL;
+ }
+
+
+ if(is_etch_client_stub(stubimpl)){
+ result = etchstub_run (thisx, stubhelper, threadpool, ((etch_client_factory*)thisx->params)->iclient, whofrom, msg);
+ }
+ else{
+
+ get_etch_session (thisx->params, session_id, &sessionparams);
+ ETCH_ASSERT (sessionparams);
+ serverimpl = sessionparams->server;
+
+ /* execute the service method execution function (stub helper),
+ * either on a thread or inline.
+ */
+ result = etchstub_run (thisx, stubhelper, threadpool, serverimpl, whofrom, msg);
+
+ }
+ } while(0);
+ /* this is the end of the line for the message */
+
+
+ return result;
+}
+
+
+/**
+ * etchstub_session_control()
+ * @param control event, caller relinquishes.
+ * @param value control value, caller relinquishes.
+ */
+int etchstub_session_control (void* data, etch_event* control, etch_object* value)
+{
+ etch_stub* thisx = (etch_stub*)data;
+ int result = -1, is_params_relinquished = FALSE;
+
+ /* a server implementation (xxxx_server_impl) can request notifications of
+ * exceptions and the like by implementing and registering i_objsession
+ * callbacks. the presence of these function pointers in the implementation
+ * serves as the indicator of whether to forward the notification.
+ */
+ i_objsession* impl_callbacks = thisx->impl_callbacks;
+
+ etch_session_control impl_session_control = impl_callbacks?
+ impl_callbacks->_session_control: NULL;
+
+ if (impl_session_control) /* if impl has requested this event, forward it */
+ if (0 == (result = impl_session_control (impl_callbacks, control, value)))
+ is_params_relinquished = TRUE;
+
+ if (0 != result)
+ { char* logmask = impl_session_control? logmask1: logmask2;
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, logmask, scstr);
+ }
+
+ if (!is_params_relinquished)
+ {
+ etch_object_destroy(control);
+ control = NULL;
+
+ etch_object_destroy(value);
+ value = NULL;
+ }
+
+ return result;
+}
+
+
+/**
+ * etch_etchstub_session_notify()
+ * @param evt event, caller relinquishes.
+ */
+int etchstub_session_notify(void* data, etch_event* evt)
+{
+ etch_stub* thisx = (etch_stub*)data;
+ const int result = etchstub_session_notify_ex (thisx, thisx->obj, evt);
+ return result;
+}
+
+
+/**
+ * etch_etchstub_session_query()
+ * @param query, caller relinquishes.
+ */
+etch_object* etchstub_session_query (void* data, etch_query* query)
+{
+ etch_stub* thisx = (etch_stub*)data;
+ etch_object* resultobj = NULL;
+ int is_params_relinquished = FALSE;
+
+ /* a server implementation (xxxx_server_impl) can request notifications of
+ * exceptions and the like by implementing and registering i_objsession
+ * callbacks. the presence of these function pointers in the implementation
+ * serves as the indicator of whether to forward the notification.
+ */
+ i_objsession* impl_callbacks = thisx->impl_callbacks;
+
+ etch_session_query impl_session_query = impl_callbacks?
+ impl_callbacks->_session_query: NULL;
+
+ if (impl_session_query) /* if impl has requested this event, forward it */
+ if (NULL != (resultobj = impl_session_query (impl_callbacks, query)))
+ is_params_relinquished = TRUE;
+
+ if (NULL == resultobj)
+ { char* logmask = impl_session_query? logmask1: logmask2;
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, logmask, sqstr);
+ }
+
+ if (!is_params_relinquished) {
+ etch_object_destroy(query);
+ query = NULL;
+ }
+
+ return resultobj;
+}
+
+
+/* - - - - - - - - - - - - - - -
+ * constructors, destructor
+ * - - - - - - - - - - - - - - -
+ */
+
+/**
+ * destroy_stub_object()
+ * stub implementation private destructor.
+ * calls back to hand-coded destructor for possible custom deallocation,
+ * then destroys the stub base, including the shared session interface.
+ */
+int destroy_stub_object (void* thisx)
+{
+ xxxx_either_stub* stubobj = NULL;
+ if (NULL == thisx) return -1;
+
+ stubobj = (xxxx_either_stub*) thisx;
+ if (!is_etch_stub(thisx)) return -1;
+
+ if (!is_etchobj_static_content(stubobj))
+ {
+ if (stubobj->destroyex) /* call back to user dtor */
+ stubobj->destroyex(thisx);
+
+ if (stubobj->stub_base)
+ etch_object_destroy(stubobj->stub_base);
+ }
+
+ return destroy_objectex((etch_object*)stubobj);
+}
+
+
+/**
+ * is_etch_stub()
+ */
+int is_etch_stub(void* x)
+{
+ int result = FALSE, objtype = x? ((etch_object*)x)->obj_type: 0;
+ switch(objtype)
+ { case ETCHTYPEB_STUB: case ETCHTYPEB_CLIENTSTUB: case ETCHTYPEB_SERVERSTUB:
+ result = TRUE;
+ }
+ return result;
+}
+
+
+/**
+ * new_stubimpl_init()
+ * generic stub implementation constructor
+ * @param implobj i_xxxx_client* or i_xxxx_server*.
+ * @param objsize number of bytes in the stub object implementation.
+ * all we know about here is the mask, or object header if you will.
+ * @param stubtype ETCH_STUBTYPE_CLIENT or ETCH_STUBTYPE_SERVER.
+ * @param userdtor a hand-coded callback in the implementation conforming
+ * to signature etch_destructor, for the purpose of freeing any custom
+ * memory allocations.
+ * @param ids the delivery service, caller retains.
+ * @param qp the queued thread pool, optional, caller retains.
+ * @param fp the free thread pool, optional, caller retains.
+ * @param params a etch_server_factory* parameter bundle, caller retains.
+ * if it is always the case that this parameter is present and is a
+ * etch_server_factory*, we can lose the ids, qp, and fp ctor parameters.
+ */
+void* new_stubimpl_init (void* implobj, const int objsize,
+ const unsigned char stubtype, etch_object_destructor userdtor,
+ i_delivery_service* ids, etch_threadpool* qp, etch_threadpool* fp,
+ void* params)
+{
+ unsigned short obj_type, class_id;
+ xxxx_either_stub* newstub = NULL;
+
+ switch(stubtype)
+ {
+ case ETCH_STUBTYPE_CLIENT:
+ obj_type = ETCHTYPEB_CLIENTSTUB;
+ class_id = CLASSID_CLIENTSTUB;
+ break;
+ case ETCH_STUBTYPE_SERVER:
+ obj_type = ETCHTYPEB_SERVERSTUB;
+ class_id = CLASSID_SERVERSTUB;
+ break;
+ default: return NULL;
+ }
+
+ newstub = (xxxx_either_stub*) new_object (objsize, obj_type, class_id);
+ ((etch_object*)newstub)->destroy = destroy_stub_object; /* private */
+ newstub->destroyex = userdtor; /* public */
+ newstub->stub_base = new_stub (implobj, stubtype, ids, qp, fp);
+ newstub->stub_base->stubobj = (etch_object*) newstub;
+ newstub->stub_base->params = params;
+
+ return newstub;
+}
+
+
+/**
+ * new_clientstub_init()
+ * generic client stub implementation constructor.
+ * @param bytelen number of bytes in the client stub object implementation.
+ * all we know about here is the mask, or object header if you will.
+ * @param dtor a hand-coded callback in the implementation conforming
+ * to signature etch_destructor, for the purpose of freeing any custom
+ * memory allocations.
+ * @param ids the delivery service, caller retains.
+ * @param qp the queued thread pool, optional, caller retains.
+ * @param fp the free thread pool, optional, caller retains.
+ * @param params a etch_server_factory* parameter bundle, caller retains.
+ * if it is always the case that this parameter is present and is a
+ * etch_server_factory*, we can lose the ids, qp, and fp ctor parameters.
+ */
+void* new_clientstub_init (void* implobj, const int bytelen, etch_object_destructor dtor,
+ i_delivery_service* ids, etch_threadpool* qp, etch_threadpool* fp, void* params)
+{
+ return new_stubimpl_init (implobj, bytelen, ETCH_STUBTYPE_CLIENT,
+ dtor, ids, qp, fp, params);
+}
+
+
+/**
+ * new_serverstub_init()
+ * generic server stub implementation constructor.
+ * @param bytelen number of bytes in the server stub object implementation.
+ * all we know about here is the mask, or object header if you will.
+ * @param dtor a hand-coded callback in the implementation conforming
+ * to signature etch_destructor, for the purpose of freeing any custom
+ * memory allocations.
+ * @param ids the delivery service, caller retains.
+ * @param qp the queued thread pool, optional, caller retains.
+ * @param fp the free thread pool, optional, caller retains.
+ * @param params a etch_server_factory* parameter bundle, caller retains.
+ * if it is always the case that this parameter is present and is a
+ * etch_server_factory*, we can lose the ids, qp, and fp ctor parameters.
+ */
+void* new_serverstub_init (void* implobj, const int bytelen, etch_object_destructor dtor,
+ i_delivery_service* ids, etch_threadpool* qp, etch_threadpool* fp, void* params)
+{
+ return new_stubimpl_init (implobj, bytelen, ETCH_STUBTYPE_SERVER,
+ dtor, ids, qp, fp, params);
+}
+
+
+
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_transportint.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_transportint.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_transportint.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/support/etch_transportint.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,130 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_transportint.c
+ * transport interface
+ */
+
+#include "etch_transportint.h"
+#include "etch_sessionint.h"
+#include "etch_mem.h"
+#include "etch_session_message.h"
+
+int etchtransport_def_transport_control (void*, etch_event*, etch_object*);
+int etchtransport_def_transport_notify (void*, etch_event*);
+etch_object* etchtransport_def_transport_query (void*, etch_query*);
+
+i_session* etchtransport_def_get_session(void* thisx);
+void etchtransport_def_set_session(void* thisx, void* session);
+
+
+/**
+ * new_default_transport_interface
+ * return a transport interface populated with defaults for virtuals.
+ * caller owns returned object, not an etch object, use etch_free() to destroy.
+ */
+i_transport* new_default_transport_interface(void* thisx)
+{
+ i_transport* newi = etch_malloc(sizeof(i_transport), ETCHTYPEB_RAWOBJECT);
+
+ newi->transport_control = etchtransport_def_transport_control;
+ newi->transport_notify = etchtransport_def_transport_notify;
+ newi->transport_query = etchtransport_def_transport_query;
+ newi->get_session = etchtransport_def_get_session;
+ newi->set_session = etchtransport_def_set_session;
+ newi->thisx = thisx;
+ return newi;
+}
+
+
+/**
+ * new_transport_interface
+ * return a transport interface populated with specified virtuals.
+ * caller owns returned object, not an etch object, use etch_free() to destroy.
+ */
+i_transport* new_transport_interface(void* thisx,
+ etch_transport_control sc, etch_transport_notify sn, etch_transport_query sq)
+{
+ i_transport* newi = new_default_transport_interface(thisx);
+
+ if (sc) newi->transport_control = sc;
+ if (sn) newi->transport_notify = sn;
+ if (sq) newi->transport_query = sq;
+ return newi;
+}
+
+
+/**
+ * new_transport_interface_ex
+ * return a transport interface populated with specified virtuals.
+ * caller owns returned object, not an etch object, use etch_free() to destroy.
+ */
+i_transport* new_transport_interface_ex(void* thisx,
+ etch_transport_control sc, etch_transport_notify sn, etch_transport_query sq,
+ etch_transport_get_session gs, etch_transport_set_session ss)
+{
+ i_transport* newi = new_transport_interface(thisx, sc, sn, sq);
+
+ if (gs) newi->get_session = gs;
+ if (ss) newi->set_session = ss;
+ return newi;
+}
+
+
+/**
+ * clone_transport()
+ */
+i_transport* clone_transport(void* thisx, const i_transport* thattransport)
+{
+ i_transport* newtransport
+ = thattransport? new_default_transport_interface(thisx): NULL;
+
+ if (newtransport)
+ memcpy(newtransport, thattransport, sizeof(i_transport));
+
+ return newtransport;
+}
+
+
+int etchtransport_def_transport_control (void* obj, etch_event* evt, etch_object* v)
+{
+ return -1;
+}
+
+int etchtransport_def_transport_notify (void* obj, etch_event* evt)
+{
+ return -1;
+}
+
+
+etch_object* etchtransport_def_transport_query (void* obj, etch_query* query)
+{
+ return NULL;
+}
+
+
+i_session* etchtransport_def_get_session(void* thisx)
+{
+ return NULL;
+}
+
+
+void etchtransport_def_set_session(void* thisx, void* session)
+{
+}
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,243 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_connection.c
+ * connection client and server classes - tcp, udp
+ */
+#include "etch.h"
+#include "etch_runtime.h"
+#include "etch_thread.h"
+#include "etch_connection.h"
+#include "etch_encoding.h"
+#include "etch_flexbuffer.h"
+#include "etch_objecttypes.h"
+#include "etch_log.h"
+#include "etch_mem.h"
+
+static const char* LOG_CATEGORY = "etch_connection";
+
+// extern types
+extern etch_pool_t* g_etch_main_pool;
+extern apr_thread_mutex_t* g_etch_main_pool_mutex;
+
+const wchar_t* ETCH_CONNECTION_RECONDELAY = L"TcpConnection.reconnect_delay";
+const wchar_t* ETCH_CONNECTION_AUTOFLUSH = L"TcpConnection.autoFlush";
+const wchar_t* ETCH_CONNECTION_KEEPALIVE = L"TcpConnection.keepAlive";
+const wchar_t* ETCH_CONNECTION_LINGERTIME = L"TcpConnection.lingerTime";
+const wchar_t* ETCH_CONNECTION_NODELAY = L"TcpConnection.noDelay";
+const wchar_t* ETCH_CONNECTION_TRAFCLASS = L"TcpConnection.trafficClass";
+const wchar_t* ETCH_CONNECTION_BUFSIZE = L"TcpConnection.bufferSize";
+const wchar_t* ETCH_TCPLISTENER_BACKLOG = L"TcpListener.backlog";
+
+unsigned next_etch_connection_id();
+
+unsigned connection_id_farm;
+
+/*
+ * is_good_tcp_params()
+ */
+int is_good_tcp_params(etch_url* url, void* resources, etch_rawsocket* socket)
+{
+ int whicherr = 0;
+
+ if (NULL == socket)
+ {
+ if (NULL == url->host) whicherr = 0x1;
+ if (url->port <= 0 || url->port > 0xffff) whicherr |= 0x2;
+ }
+
+ return whicherr == 0;
+}
+
+
+/*
+ * etch_socket_set_timeout
+ * set timeout for specified socket
+ */
+int etch_socket_set_timeout(etch_rawsocket* socket, const unsigned ms)
+{
+ const int result = apr_socket_timeout_set(socket, ms * 1000);
+ return 0 == result? 0: -1;
+}
+
+
+/*
+ * etch_socket_get_timeout
+ * get timeout set on specified socket
+ */
+int etch_socket_get_timeout(etch_rawsocket* socket, unsigned* retms)
+{
+ int64 val = 0;
+ if (0 != apr_socket_timeout_get(socket, &val)) return -1;
+ if (retms) *retms = (unsigned) (val / 1000);
+ return 0;
+}
+
+
+/**
+ * etchconx_set_socket_options()
+ */
+int etchconx_set_socket_options(void *c)
+{
+ return 0;
+}
+
+
+/**
+ * next_etch_connection_id()
+ * return a unique ID used to identify a connection instance
+ */
+unsigned next_etch_connection_id()
+{
+ do { apr_atomic_inc32 ((volatile apr_uint32_t*) &connection_id_farm);
+ } while(connection_id_farm == 0);
+
+ return connection_id_farm;
+}
+
+
+/**
+ * etch_defconx_on_data()
+ * default receive data handler
+ */
+int etch_defconx_on_data (void* conx, const int unused, int length, void* voidData)
+{
+ char* msgmask = "connxn %d no data handler provided for %d bytes\n";
+ etch_connection* cx = (etch_connection*) conx;
+ ETCH_ASSERT(is_etch_connection(cx));
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN, msgmask, cx->conxid, length);
+ return 0;
+}
+
+
+/*
+ * etch_defconx_on_event()
+ * default handler for connection events
+ */
+int etch_defconx_on_event (void* c, const int e, int p1, void* p2)
+{
+ return 0;
+}
+
+/*
+ * etchconx_wait_up()
+ * block until connection comes up or timeout
+ */
+int etchconx_wait_up(etch_connection* cx, int timeoutms)
+{
+ etch_status_t status = ETCH_SUCCESS;
+ status = etch_wait_timedwait(cx->wait_up, ETCH_CONXEVT_UP, timeoutms);
+ if(status != ETCH_SUCCESS) {
+ return -1;
+ }
+ return 0;
+}
+
+/*
+ * etchconx_wait_down()
+ * block until connection is closed or timeout
+ */
+int etchconx_wait_down(etch_connection* cx, int timeoutms)
+{
+ etch_status_t status = ETCH_SUCCESS;
+ status = etch_wait_timedwait(cx->wait_down, ETCH_CONXEVT_DOWN, timeoutms);
+ if(status != ETCH_SUCCESS) {
+ return -1;
+ }
+ return 0;
+}
+
+/*
+ * etch_destroy_connection()
+ * free memory owned by connection.
+ */
+int etch_destroy_connection(etch_connection* cx)
+{
+ if (cx)
+ {
+ etch_wait_destroy(cx->wait_up);
+ etch_wait_destroy(cx->wait_down);
+ etch_mutex_destroy(cx->mutex);
+ etch_free(cx->hostname);
+ /* free apr subpool */
+ //printf("5 destroying apr pool %p\n",cx->aprpool);
+ apr_pool_destroy(cx->aprpool);
+
+ }
+ return 0;
+}
+
+/*
+ * etch_init_connection()
+ * initialize an etch_connection (abstract base)
+ */
+int etch_init_connection (etch_connection* cx, etch_rawsocket* socket, void* owner)
+{
+ etch_status_t status = ETCH_SUCCESS;
+ memset(cx, 0, sizeof(etch_connection));
+
+ cx->conxid = next_etch_connection_id();
+ cx->owner = owner;
+
+ // TODO: pool handling
+
+ /* set memory pool used for APR objects associated with this connection.
+ * if the connection is being created with an existing socket, use the
+ * memory pool already assigned to it, otherwise partition a new subpool.
+ * if connection allocates a subpool, connection will free it on destroy.
+ */
+ if (socket) {
+ cx->aprpool = apr_socket_pool_get(socket);
+ }
+
+ if (cx->aprpool == NULL) {
+ // TODO: pool
+ apr_pool_create(&cx->aprpool, g_etch_main_pool);
+ //printf("3 creating apr pool %p\n",cx->aprpool);
+ /* set pool abort function */
+ apr_pool_abort_set(etch_runtime_mem_abort, cx->aprpool);
+
+ cx->is_ownpool = TRUE;
+ }
+
+ if (cx->aprpool == NULL)
+ { cx->aprpool = g_etch_main_pool;
+ cx->is_ownpool = FALSE;
+ }
+
+ cx->sig = ETCH_CONX_SIG;
+
+ // TODO: pool
+ status = etch_mutex_create(&cx->mutex, ETCH_MUTEX_NESTED, NULL);
+ ETCH_ASSERT(status == ETCH_SUCCESS);
+ // TODO: pool
+ status = etch_wait_create(&cx->wait_up, NULL);
+ ETCH_ASSERT(status == ETCH_SUCCESS);
+ // TODO: pool
+ status = etch_wait_create(&cx->wait_down, NULL);
+ ETCH_ASSERT(status == ETCH_SUCCESS);
+
+ cx->bufsize = ETCH_CONX_DEFAULT_BUFSIZE;
+ cx->on_event = etch_defconx_on_event;
+ cx->on_data = etch_defconx_on_data;
+ cx->set_socket_options = etchconx_set_socket_options;
+ return 0;
+}
+
+
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection_event.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection_event.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection_event.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_connection_event.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,336 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_conxevent.c
+ * event handler overrides for the various connection superclasses
+ */
+
+#include "etch_thread.h"
+#include "etch_tcp_connection.h"
+#include "etch_tcp_server.h"
+#include "etch_objecttypes.h"
+#include "etch_log.h"
+
+static const char* LOG_CATEGORY = "etch_connection_event";
+
+/*
+ * etch_deftcplistener_on_event()
+ * default handler for listener events
+ */
+int etch_deftcplistener_on_event(etch_tcp_server* l, etch_tcp_connection* c, const int e, int p1, void* p2)
+{
+ char cxstr[24], *smask = "server %u";
+
+ if (c)
+ { c->cx.listener = (etch_object*) l;
+ return c->cx.on_event(c, e, p1, p2);
+ }
+
+ apr_snprintf(cxstr, sizeof(cxstr), smask, l->listener_id);
+
+ switch(e)
+ {
+ case ETCH_CONXEVT_CREATED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s created\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_CREATERR:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s not created\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_OPENING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s opening ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_DESTROYING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s destroying ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_DESTROYED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s destroyed\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_SHUTDOWN:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN, "%s shutdown request detected\n", cxstr);
+ break;
+ }
+
+ return 0;
+}
+
+
+/*
+ * etch_tcpconx_on_event()
+ * default handler for connection events
+ */
+int etch_tcpconx_on_event(void* data, const int e, int p1, void* p2)
+{
+ etch_tcp_connection* c = (etch_tcp_connection*)data;
+ int result = 0, lid = 0;
+ char cxstr[32], estr[128];
+ char *scmask = "server %u connxn %u", *cmask = "connxn %u";
+
+ if (is_etch_tcpserver(c->cx.listener))
+ lid = ((etch_tcp_server*)(c->cx.listener))->listener_id;
+ if (lid)
+ sprintf(cxstr, scmask, lid, c->cx.conxid);
+ else sprintf(cxstr, cmask, c->cx.conxid);
+
+ switch(e)
+ {
+ case ETCH_CONXEVT_RECEIVING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_XDEBUG, "%s begin receive (block) ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_RECEIVERR:
+ apr_strerror(p1, estr, 128);
+ if(IS_ETCH_SOCKET_TIMEOUT(p1))
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s socket receive timed out\n", cxstr);
+ else
+ if(p1 == APR_OTHER_END_CLOSED)
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN,"%s connection was broken\n", cxstr);
+ else
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s apr_socket_recv() %s\n", cxstr, estr);
+ break;
+
+ case ETCH_CONXEVT_RECEIVED: /* "eod=%d", p1 */
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s end receive %d bytes\n", cxstr, (int)(size_t)p2);
+ break;
+
+ case ETCH_CONXEVT_CONXCLOSED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s local connection closed\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_PEERCLOSED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s peer connection closed\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_RECEIVEND:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_XDEBUG, "%s exit receive loop\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_SENDING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_XDEBUG, "%s begin send\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_SENDERR:
+ apr_strerror(p1, estr, 128);
+ if(IS_ETCH_SOCKET_TIMEOUT(p1))
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s socket send timed out\n", cxstr);
+ else
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s apr_socket_send() %s\n", cxstr, estr);
+ break;
+
+ case ETCH_CONXEVT_SENDEND:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s end send %d bytes\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_RCVPUMP_START:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s start receive pump\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_RCVPUMP_RECEIVING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s thread %d receiving ...\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_RCVPUMP_ERR:
+ result = -1;
+ if(p1 == APR_OTHER_END_CLOSED)
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN, "%s connection was broken\n", cxstr);
+ else
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s receive failed\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_RCVPUMP_STOP:
+ result = p1;
+ if (result >= 0)
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s receive pump on thread %d exited\n", cxstr, (int) (size_t) p2);
+ else
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s receive pump abnormal exit\n", cxstr);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s destroying accepted connection\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_ACCEPTING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s accepting ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_ACCEPTERR:
+ result = -1;
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s apr_socket_accept() %s\n", cxstr, estr);
+ break;
+
+ case ETCH_CONXEVT_ACCEPTED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s accepted\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_CREATED:
+ if(p1)
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s created for socket %x\n", cxstr, p1);
+ else
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s created\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_CREATERR:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s not created\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_OPENING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s opening ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_OPENERR:
+ result = -1;
+
+ switch(p1)
+ {
+ case 0:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s not opened\n", cxstr);
+ break;
+ case 2:
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN, "%s socket connect: %s\n", cxstr, estr);
+ break;
+ case 3:
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR,
+ "%s socket create: %s\n", cxstr, estr);
+ break;
+ case 4:
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR,
+ "%s sockaddr info: %s\n", cxstr, estr);
+ break;
+ case 5:
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR,
+ "%s socket bind: %s\n", cxstr, estr);
+ break;
+ case 6:
+ apr_strerror((apr_status_t)(size_t)p2, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s socket listen: %s\n", cxstr, estr);
+ break;
+ }
+ break;
+
+ case ETCH_CONXEVT_OPENED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s opened\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_STARTING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s starting ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_STARTERR:
+ result = -1;
+ switch(p1)
+ {
+ case 0:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s not started\n", cxstr);
+ break;
+ case 1:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s etch_threadpool.run()\n", cxstr);
+ break;
+ }
+ break;
+
+ case ETCH_CONXEVT_STARTED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s started\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_ACCEPTPUMPEXIT:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s accept pump on thread %d exited\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_ACCEPTPUMPEXITERR:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s accept pump abnormal exit\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_LISTENED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s server id is %d\n", cxstr, p1);
+ break;
+
+ case ETCH_CONXEVT_UP:
+ case ETCH_CONXEVT_DOWN:
+ break;
+
+ case ETCH_CONXEVT_SOCKOPTERR:
+ result = 1; /* to increment counter */
+ apr_strerror(p1, estr, 128);
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s set socket option %s: %s\n", cxstr, p2, estr);
+ break;
+
+ case ETCH_CONXEVT_SHUTDOWN:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_WARN, "%s shutdown request detected\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_STOPPING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s stopping ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_STOPERR:
+ result = -1;
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s not stopped\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_STOPPED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s stopped\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_CLOSING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s closing ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_CLOSERR:
+ result = -1;
+
+ switch(p1)
+ {
+ case 0:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s not closed\n", cxstr);
+ break;
+ case 1:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s close when not open\n", cxstr);
+ break;
+ case 2:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR, "%s concurrent close denied\n", cxstr);
+ break;
+ case 3:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR,
+ "%s apr_socket_close() error %d\n", cxstr, (int)(size_t)p2);
+ break;
+ }
+ break;
+
+ case ETCH_CONXEVT_CLOSED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s closed\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_DESTROYING:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s destroying ...\n", cxstr);
+ break;
+
+ case ETCH_CONXEVT_DESTROYED:
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "%s destroyed\n", cxstr);
+ break;
+ }
+
+ return result;
+}
+
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_mailbox_manager.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_mailbox_manager.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_mailbox_manager.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_mailbox_manager.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,120 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_mailboxmgr.c
+ * i_mailbox manager interface
+ */
+
+#include "etch_mailbox_manager.h"
+#include "etch_message.h"
+#include "etch_objecttypes.h"
+#include "etch_log.h"
+/*
+static const char* LOG_CATEGORY = "etch_mailbox_manager";
+*/
+
+int destroy_mailbox_manager_interface(void*);
+int etchmailboxmgr_def_headersize(i_mailbox_manager*);
+int etchmailboxmgr_def_transport_call(i_mailbox_manager*, etch_who*, etch_message*, i_mailbox**);
+int etchmailboxmgr_def_redeliver (void*, etch_who*, etch_message*);
+int etchmailboxmgr_def_unregister(void*, i_mailbox*);
+
+
+/**
+ * new_mailboxmgr_interface()
+ * i_mailbox_manager constructor.
+ * @param thisx the mailbox manager object.
+ * @param itm transportmesssage interface, caller retains, can be null.
+ * @param ism sessionmessage interface, caller retains, can be null.
+ */
+i_mailbox_manager* new_mailboxmgr_interface(void* thisx,
+ i_transportmessage* itm, i_sessionmessage* ism)
+{
+ i_mailbox_manager* newi = (i_mailbox_manager*) new_object(sizeof(i_mailbox_manager), ETCHTYPEB_MBOX_MANAGER, CLASSID_MBOX_MANAGER);
+
+ newi->thisx = thisx;
+ ((etch_object*)newi)->clone = clone_null;
+ ((etch_object*)newi)->destroy = destroy_mailbox_manager_interface;
+
+ newi->transport_call = (etch_mbm_transport_call)etchmailboxmgr_def_transport_call;
+ newi->redeliver = etchmailboxmgr_def_redeliver;
+ newi->unregister = etchmailboxmgr_def_unregister;
+
+ newi->itm = itm? itm: new_transportmsg_interface(newi, NULL, NULL);
+ newi->transport_message = newi->itm->transport_message;
+ newi->transport_message = newi->itm->transport_message;
+ newi->transport_control = newi->itm->transport_control;
+ newi->transport_notify = newi->itm->transport_notify;
+ newi->transport_query = newi->itm->transport_query;
+ newi->get_session = newi->itm->get_session;
+ newi->set_session = newi->itm->set_session;
+
+ return newi;
+}
+
+
+/**
+ * destroy_mailbox_manager_interface()
+ * i_mailbox_manager destructor
+ */
+int destroy_mailbox_manager_interface(void* data)
+{
+ i_mailbox_manager* mgr = (i_mailbox_manager*)data;
+ if (NULL == mgr) return -1;
+
+ if (!is_etchobj_static_content(mgr))
+ {
+ etch_object_destroy(mgr->ism);
+
+ etch_object_destroy(mgr->itm);
+ }
+
+ return destroy_objectex((etch_object*)mgr);
+}
+
+
+/**
+ * etchmailboxmgr_def_transport_call()
+ * default implementation of i_mailbox_manager::transport_call()
+ */
+int etchmailboxmgr_def_transport_call(i_mailbox_manager* imbm, etch_who* whoto, etch_message* msg, i_mailbox** out)
+{
+ return -1;
+}
+
+
+/**
+ * etchmailboxmgr_def_redeliver()
+ * default implementation of i_mailbox_manager::redeliver()
+ */
+int etchmailboxmgr_def_redeliver(void* mailbox, etch_who* whofrom, etch_message* msg)
+{
+ return -1;
+}
+
+
+/**
+ * etchmailboxmgr_def_unregister()
+ * default implementation of i_mailbox_manager::unregister()
+ */
+int etchmailboxmgr_def_unregister(void* imbm, i_mailbox* mbox)
+{
+ return -1;
+}
+
Added: incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_messagizer.c
URL: http://svn.apache.org/viewvc/incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_messagizer.c?rev=966314&view=auto
==============================================================================
--- incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_messagizer.c (added)
+++ incubator/etch/branches/etch-c/binding-c/runtime/c/src/main/transport/etch_messagizer.c Wed Jul 21 16:58:23 2010
@@ -0,0 +1,470 @@
+/* $Id$
+ *
+ * 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.
+ */
+
+/*
+ * etch_msgizer.c
+ * messagizer accepts packets and translates them to messages,
+ * and it accepts messages and translates them to packets.
+ */
+
+#include "etch_messagizer.h"
+#include "etch_tdformat.h"
+#include "etch_thread.h"
+#include "etch_objecttypes.h"
+#include "etch_log.h"
+
+static const char* LOG_CATEGORY = "etch_messagizer";
+
+// extern types
+extern apr_pool_t* g_etch_main_pool;
+
+int destroy_messagizer(void*);
+int etch_msgizer_transport_message(void*, void*, void*);
+int etch_msgizer_session_packet (void*, void*, void*);
+int etch_msgizer_session_control (void*, etch_event*, etch_object*);
+int etch_msgizer_session_notify (void*, etch_event*);
+etch_object* etch_msgizer_session_query (void*, etch_query*);
+int etch_msgizer_transport_control(void*, etch_event*, etch_object*);
+int etch_msgizer_transport_notify (void*, etch_event*);
+etch_object* etch_msgizer_transport_query (void*, etch_query*);
+
+
+/* - - - - - - - - - - - - - - -
+ * etch_messagizer
+ * - - - - - - - - - - - - - - -
+ */
+
+/**
+ * new_messagizer()
+ * etch_messagizer public constructor
+ * @param ixport transport interface owned by caller
+ * @param uri a URI string owned by caller.
+ * @param resx a resources map owned by caller.
+ */
+
+etch_messagizer* new_messagizer (i_transportpacket* ixp, wchar_t* uri, etch_resources* resx)
+{
+ etch_url* url = new_url(uri);
+
+ etch_messagizer* messagizer = new_messagizer_a(ixp, url, resx);
+
+ etch_object_destroy(url);
+ return messagizer;
+}
+
+
+/**
+ * new_messagizer_a()
+ * etch_messagizer private constructor
+ * @param ipacket transport interface owned by caller
+ * @param uri a URI string owned by caller.
+ * @param resxmap a resources map owned by caller.
+ */
+etch_messagizer* new_messagizer_a (i_transportpacket* ixp, etch_url* url, etch_resources* resxmap)
+{
+ etch_status_t status = ETCH_SUCCESS;
+ tagdata_format_factory* formatfactory = NULL;
+ etch_messagizer* messagizer = NULL;
+ i_transport* itransport = NULL;
+ etch_value_factory* vf = NULL;
+ etch_string* formatobj = NULL;
+ i_session* isession = NULL;
+ etch_mutex* mutex = NULL;
+ int result = -1;
+
+ do
+ {
+ formatobj = (etch_string*) etchurl_get_term (url, ETCH_RESXKEY_MSGIZER_FORMAT);
+
+ if (NULL == formatobj && NULL != resxmap)
+ formatobj = (etch_string*) etch_resources_get (resxmap, ETCH_RESXKEY_MSGIZER_FORMAT);
+
+ if (NULL == formatobj) {
+ ETCH_LOGW(LOG_CATEGORY, ETCH_LOG_ERROR, L"URI missing format specifier '%s'\n", ETCH_RESXKEY_MSGIZER_FORMAT);
+ break;
+ }
+
+ formatfactory = get_format_factory(formatobj->v.valw);
+ if (NULL == formatfactory) {
+ ETCH_LOGW(LOG_CATEGORY, ETCH_LOG_ERROR, L"no format '%s' is installed\n",
+ formatobj->v.valw);
+ break;
+ }
+
+ vf = resxmap? (etch_value_factory*) etch_resources_get(resxmap,
+ ETCH_RESXKEY_MSGIZER_VALUFACT): NULL;
+ if (NULL == vf)
+ {
+ ETCH_LOGW(LOG_CATEGORY, ETCH_LOG_ERROR, L"no value factory '%s' is installed\n", ETCH_RESXKEY_MSGIZER_VALUFACT);
+ break;
+ }
+
+ // TODO: pool
+ status = etch_mutex_create(&mutex, ETCH_MUTEX_NESTED, NULL);
+ if(status != ETCH_SUCCESS) {
+ // error
+ break;
+ }
+
+ /* - - - - - - - - - - - - - - -
+ * etch_messagizer
+ * - - - - - - - - - - - - - - -
+ */
+ messagizer = (etch_messagizer*) new_object(sizeof(etch_messagizer), ETCHTYPEB_MESSAGIZER, CLASSID_MESSAGIZER);
+
+ ((etch_object*)messagizer)->destroy = destroy_messagizer;
+ ((etch_object*)messagizer)->clone = clone_null;
+ messagizer->msglock = mutex;
+ messagizer->msgbuf = new_flexbuffer(ETCH_DEFSIZE); /* 2K default */
+
+ /* set our transport to that of next lower layer (packetizer) */
+ messagizer->transport = ixp; /* i_transportpacket owned by caller */
+
+ /* - - - - - - - - - - - - - - -
+ * i_transportmessage
+ * - - - - - - - - - - - - - - -
+ */
+ itransport = new_transport_interface_ex (messagizer,
+ (etch_transport_control) etch_msgizer_transport_control,
+ (etch_transport_notify) etch_msgizer_transport_notify,
+ (etch_transport_query) etch_msgizer_transport_query,
+ etch_msgizer_get_session,
+ etch_msgizer_set_session);
+
+ /* instantiate i_transportmessage interface which messagizer implements */
+ messagizer->transportmsg = new_transportmsg_interface(messagizer,
+ etch_msgizer_transport_message,
+ itransport); /* transportmsg now owns itransport */
+
+ /* copy i_transportmessage interface methods up to messagizer */
+ messagizer->transport_message = etch_msgizer_transport_message;
+ messagizer->transport_control = itransport->transport_control;
+ messagizer->transport_notify = itransport->transport_notify;
+ messagizer->transport_query = itransport->transport_query;
+ messagizer->get_session = itransport->get_session;
+ messagizer->set_session = itransport->set_session;
+
+ /* - - - - - - - - - - - - - - -
+ * i_sessionpacket
+ * - - - - - - - - - - - - - - -
+ */
+ isession = new_session_interface(messagizer,
+ (etch_session_control) etch_msgizer_session_control,
+ (etch_session_notify) etch_msgizer_session_notify,
+ (etch_session_query) etch_msgizer_session_query);
+
+ /* instantiate i_sessionpacket interface which messagizer implements */
+ messagizer->sessionpkt = new_sessionpkt_interface(messagizer,
+ etch_msgizer_session_packet,
+ isession); /* transportmsg now owns isession */
+
+ /* copy session interface to parent */
+ messagizer->session_packet = etch_msgizer_session_packet;
+ messagizer->session_control = isession->session_control;
+ messagizer->session_notify = isession->session_notify;
+ messagizer->session_query = isession->session_query;
+
+ /* finally set session of next lower layer to our session */
+ /* fyi we must pass the implementor of transport as thisx, i.e. packetizer */
+ messagizer->transport->set_session (messagizer->transport->thisx, messagizer->sessionpkt);
+
+ /* - - - - - - - - - - - - - - -
+ * tagged data in out
+ * - - - - - - - - - - - - - - -
+ */
+ messagizer->tdi = formatfactory->new_tagdata_input(vf);
+ if (NULL == messagizer->tdi) break;
+
+ messagizer->tdo = formatfactory->new_tagdata_output(vf);
+ if (NULL == messagizer->tdo) break;
+
+ result = 0;
+
+ } while(0);
+
+
+ if (-1 == result) {
+ etch_object_destroy(vf);
+ etch_object_destroy(messagizer);
+ messagizer = NULL;
+ }
+
+ etch_object_destroy(formatfactory);
+
+ return messagizer;
+}
+
+
+/**
+ * destroy_messagizer()
+ * destructor for etch_messagizer
+ */
+int destroy_messagizer (void* data)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ if (!is_etchobj_static_content(thisx))
+ {
+ etch_object_destroy(thisx->transportmsg);
+ etch_object_destroy(thisx->sessionpkt);
+ etch_object_destroy(thisx->tdi);
+ etch_object_destroy(thisx->tdo);
+ etch_object_destroy(thisx->msgbuf);
+ etch_object_destroy(thisx->msglock);
+
+ }
+
+ return destroy_objectex((etch_object*) thisx);
+}
+
+
+/**
+ * etch_msgizer_get_transport()
+ * @return a reference to the messagizer transport interface.
+ * this object is owned by whatever object created the messagizer.
+ */
+i_transportpacket* etch_msgizer_get_transport (etch_messagizer* thisx)
+{
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ return thisx->transport;
+}
+
+
+/* - - - - - - - - - - - - - - -
+ * i_transportmessage
+ * - - - - - - - - - - - - - - -
+ */
+
+/**
+ * etch_msgizer_transport_message()
+ * i_transportmessage::transport_message override.
+ * serializes message and delivers its data to transport
+ * @param whoto recipient - caller retains this memory, can be null.
+ * @param message the message
+ * caller relinquishes this memory on success, retains on failure.
+ * @return 0 success, -1 error.
+ */
+int etch_msgizer_transport_message (void* data, void* whoData, void* messageData)
+{
+
+ etch_who* whoto = (etch_who*)whoData;
+ etch_message* msg = (etch_message*)messageData;
+
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ etch_status_t status = ETCH_SUCCESS;
+ int result = 0;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+
+ status = etch_mutex_lock(thisx->msglock);
+ ETCH_ASSERT(status == ETCH_SUCCESS);
+
+ do
+ { const int headersize = thisx->transport->get_headersize (thisx->transport);
+
+ etch_flexbuf_skip (thisx->msgbuf, headersize, TRUE);
+
+ /* serialize the message to the buffer */
+ result = ((struct i_tagged_data_output*)
+ (((etch_object*)thisx->tdo)->vtab))
+ ->write_message (thisx->tdo, msg, thisx->msgbuf);
+
+ if (-1 == result) {
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_ERROR,"serialization of message to buffer failed");
+ etch_flexbuf_reset(thisx->msgbuf);
+ break;
+ }
+
+ etch_flexbuf_set_index (thisx->msgbuf, 0);
+
+
+ /* deliver packet buffer to transport */ /* msgizer->itp->pktizer */
+ result = thisx->transport->transport_packet (thisx->transport->thisx, whoto, thisx->msgbuf);
+
+ if (0 == result) {
+ etch_object_destroy(msg);
+ }
+
+ } while(0);
+
+ status = etch_mutex_unlock(thisx->msglock);
+ ETCH_ASSERT(status == ETCH_SUCCESS);
+
+ return result;
+}
+
+
+/**
+ * etch_msgizer_get_session()
+ * @return a reference to the messagizer i_sessionmessage interface.
+ * this object is owned by whatever object set the messagizer session.
+ */
+i_session* etch_msgizer_get_session (void* data)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ return (i_session*)thisx->session;
+}
+
+
+/**
+ * etch_msgizer_set_session()
+ * @param session an i_sessionmessage reference. caller owns this object.
+ */
+void etch_msgizer_set_session (void* data, void* sessionData)
+{
+ i_session* session = (i_session*)sessionData;
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ ETCH_ASSERT(is_etch_sessionmsg(session));
+ thisx->session = (i_sessionmessage*)session;
+}
+
+
+/**
+ * etch_msgizer_transport_control()
+ * i_transportmessage::transport_control override.
+ * @param control event, caller relinquishes.
+ * @param value control value, caller relinquishes.
+ */
+int etch_msgizer_transport_control (void* data, etch_event* control, etch_object* value)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ /* mzr itp mzr itp packetizer */
+ return thisx->transport->transport_control (thisx->transport->thisx, control, value);
+}
+
+
+/**
+ * etch_msgizer_transport_notify()
+ * i_transportmessage::transport_notify override.
+ * @param evt, caller relinquishes.
+ */
+int etch_msgizer_transport_notify (void* data, etch_event* evt)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ return thisx->transport->transport_notify (thisx->transport->thisx, evt);
+}
+
+
+/**
+ * etch_msgizer_transport_query()
+ * i_transportmessage::transport_query override.
+ * @param query, caller relinquishes.
+ */
+etch_object* etch_msgizer_transport_query (void* data, etch_query* query)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+ return thisx->transport->transport_query (thisx->transport->thisx, query);
+}
+
+
+/* - - - - - - - - - - - - - - -
+ * i_sessionpacket
+ * - - - - - - - - - - - - - - -
+ */
+
+/**
+ * etch_msgizer_session_packet()
+ * i_sessionpacket::session_packet override.
+ * delivers data to the session from the transport.
+ * @param from from who sent the packet.
+ * caller retains memory for this object, can be null.
+ * @param buffer the packet from the packet source.
+ * caller retains memory for this object.
+ * @return 0 success, -1 error (exception condition)
+ */
+int etch_msgizer_session_packet (void* data, void* whoData, void* bufferData)
+{
+
+ etch_who* whofrom = (etch_who*)whoData;
+ etch_flexbuffer* fbuf = (etch_flexbuffer*)bufferData;
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ etch_message* msg = NULL;
+ int is_message_handled = FALSE;
+ ETCH_ASSERT(is_etch_messagizer(thisx));
+
+ /* create an etch message from the packetized data */
+ msg = ((struct i_tagged_data_input*)((etch_object*)thisx->tdi)->vtab)->read_message (thisx->tdi, fbuf);
+
+ /* send the new message up via session.
+ * memory management rules are: if session_message() handles the message,
+ * it owns msg memory. otherwise (if not handled), msg memory is owned by
+ * the unwanted_message wrapper created here, which itself is owned by the
+ * session_notify() destination.
+ */
+ is_message_handled = (0 == thisx->session->session_message (thisx->session->thisx, whofrom, msg));
+
+ /* if the message was not handled, e.g. the message is an exception returned
+ * from a one-way message and there was therefore no mailbox to receive it,
+ * forward the message to the session via mailbox manager.
+ */
+ if (!is_message_handled) {
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "unable to post message to a mailbox\n");
+ ETCH_LOG(LOG_CATEGORY, ETCH_LOG_DEBUG, "deferring '%s' to session\n", message_aname(msg));
+ thisx->session->session_notify(thisx->session->thisx, (etch_event*)new_unwanted_message(whofrom, msg));
+ //etch_object_destroy(msg);
+ return -1;
+ }
+
+ /* regardless we have relinquished msg at this point. it is now either queued up
+ * in a mailbox, or forwarded in the unwanted message wrapper above. */
+ return 0;
+}
+
+
+/**
+ * etch_msgizer_session_control()
+ * i_sessionpacket::session_control override.
+ * @param control event, caller relinquishes.
+ * @param value control value, caller relinquishes.
+ */
+int etch_msgizer_session_control (void* data, etch_event* control, etch_object* value)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx)); /* ism mbmgr */
+ return thisx->session->session_control (thisx->session->thisx, control, value);
+}
+
+
+/**
+ * etch_msgizer_session_notify()
+ * i_sessionpacket::session_notify override.
+ * @param event, caller relinquishes.
+ */
+int etch_msgizer_session_notify (void* data, etch_event* evt)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx)); /* ism mbmgr */
+ return thisx->session->session_notify (thisx->session->thisx, evt);
+}
+
+
+/**
+ * etch_msgizer_session_query()
+ * i_sessionpacket::session_query override.
+ * @param query, caller relinquishes.
+ */
+etch_object* etch_msgizer_session_query (void* data, etch_query* query)
+{
+ etch_messagizer* thisx = (etch_messagizer*)data;
+ ETCH_ASSERT(is_etch_messagizer(thisx)); /* ism mbmgr */
+ return thisx->session->session_query (thisx->session->thisx, query);
+}
+