OpenSync: /var/data/www/opensync/code/opensync/client/opensync_client

00001 /*
00002  * libosengine - A synchronization engine for the opensync framework
00003  * Copyright (C) 2004-2005  Armin Bauer <armin.bauer@opensync.org>
00004  * 
00005  * This library is free software; you can redistribute it and/or
00006  * modify it under the terms of the GNU Lesser General Public
00007  * License as published by the Free Software Foundation; either
00008  * version 2.1 of the License, or (at your option) any later version.
00009  * 
00010  * This library is distributed in the hope that it will be useful,
00011  * but WITHOUT ANY WARRANTY; without even the implied warranty of
00012  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
00013  * Lesser General Public License for more details.
00014  * 
00015  * You should have received a copy of the GNU Lesser General Public
00016  * License along with this library; if not, write to the Free Software
00017  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
00018  * 
00019  */
00020  
00021 #include "opensync.h"
00022 #include "opensync_internals.h"
00023 
00024 #include "opensync-data.h"
00025 
00026 #include "opensync-ipc.h"
00027 #include "ipc/opensync_message_internals.h"
00028 #include "ipc/opensync_serializer_internals.h"
00029 #include "ipc/opensync_queue_internals.h"
00030 #include "plugin/opensync_objtype_sink_internals.h"
00031 
00032 #include "opensync-group.h"
00033 #include "opensync-plugin.h"
00034 #include "opensync-merger.h"
00035 #include "opensync-format.h"
00036 
00037 #include "opensync-version.h"
00038 #include "version/opensync_version_internals.h"
00039 
00040 #include "opensync-client.h"
00041 #include "opensync_client_internals.h"
00042 
00043 #ifndef _WIN32
00044 #include <sys/types.h>
00045 #include <sys/wait.h>
00046 #include <signal.h>
00047 #endif /* _WIN32 */
00048 
00049 #ifdef _WIN32
00050 /* Historical signals specified by POSIX. */
00051 #define SIGKILL 9       /* Kill (cannot be blocked, caught, or ignored). */
00052 #define SIGTERM 15      /* can be caught and interpreted or ignored by the process */
00053 typedef int pid_t;
00054 #endif //_WIN32
00055 
00056 #include "opensync_client_proxy_internals.h"
00057 #include "opensync_client_proxy_private.h"
00058 
00059 typedef struct callContext {
00060         OSyncClientProxy *proxy;
00061         
00062         initialize_cb init_callback;
00063         void *init_callback_data;
00064         
00065         finalize_cb fin_callback;
00066         void *fin_callback_data;
00067         
00068         discover_cb discover_callback;
00069         void *discover_callback_data;
00070         
00071         connect_cb connect_callback;
00072         void *connect_callback_data;
00073 
00074         connect_done_cb connect_done_callback;
00075         void *connect_done_callback_data;
00076         
00077         disconnect_cb disconnect_callback;
00078         void *disconnect_callback_data;
00079 
00080         read_cb read_callback;
00081         void *read_callback_data;
00082         
00083         get_changes_cb get_changes_callback;
00084         void *get_changes_callback_data;
00085         
00086         commit_change_cb commit_change_callback;
00087         void *commit_change_callback_data;
00088         
00089         committed_all_cb committed_all_callback;
00090         void *committed_all_callback_data;
00091         
00092         sync_done_cb sync_done_callback;
00093         void *sync_done_callback_data;
00094 } callContext;
00095 
00096 //portable kill pid helper
00097 /*static int _osync_kill(pid_t pid, int sig) 
00098         {
00099         #ifndef _WIN32
00100         return kill(pid, sig);
00101         #else //_WIN32
00102         int ret = -1;
00103         DWORD dwExitCode = 0;
00104 
00105         HANDLE hProc = OpenProcess(1, 0, pid);
00106         GenerateConsoleCtrlEvent(CTRL_C_EVENT, pid);
00107         WaitForSingleObject(hProc, 3000);
00108         GetExitCodeProcess(hProc, &dwExitCode);
00109         
00110         if(dwExitCode != STILL_ACTIVE) {
00111         ret = 0;
00112         goto end;
00113         }
00114         if (sig == SIGKILL) {
00115         if (TerminateProcess(hProc, 0))
00116         ret = 0;
00117         else
00118         ret = -1;
00119         }
00120         end:
00121         CloseHandle(hProc);
00122         return (ret);
00123         #endif //_WIN32
00124         }*/
00125 
00126 /*static char *_osync_client_pid_filename(OSyncClientProxy *proxy)
00127         {
00128         return g_strdup_printf("%s%cosplugin.pid", proxy->path, G_DIR_SEPARATOR);
00129         }*/
00130 
00131 /*static osync_bool osync_client_remove_pidfile(OSyncClientProxy *proxy, OSyncError **error)
00132         {
00133         char *pidpath = _osync_client_pid_filename(proxy);
00134 
00135         if (unlink(pidpath) < 0) {
00136         osync_error_set(error, OSYNC_ERROR_GENERIC, "Couldn't remove pid file: %s", g_strerror(errno));
00137         g_free(pidpath);
00138         return FALSE;
00139         }
00140         
00141         g_free(pidpath);
00142         return TRUE;
00143         }*/
00144 
00145 /*static osync_bool _osync_client_create_pidfile(OSyncClientProxy *proxy, OSyncError **error)
00146         {
00147         char *pidpath = _osync_client_pid_filename(proxy);
00148         char *pidstr = g_strdup_printf("%ld", (long)proxy->child_pid);
00149 
00150         if (!osync_file_write(pidpath, pidstr, strlen(pidstr), 0644, error)) {
00151         g_free(pidstr);
00152         g_free(pidpath);
00153         return FALSE;
00154         }
00155 
00156         g_free(pidstr);
00157         g_free(pidpath);
00158         return TRUE;
00159         }*/
00160 
00161 /*static osync_bool _osync_client_kill_old_osplugin(OSyncClientProxy *proxy, OSyncError **error)
00162         {
00163         osync_bool ret = FALSE;
00164 
00165         char *pidstr;
00166         unsigned int pidlen;
00167         pid_t pid;
00168 
00169         char *pidpath = _osync_client_pid_filename(proxy);
00170 
00171         // Simply returns if there is no PID file
00172         if (!g_file_test(pidpath, G_FILE_TEST_EXISTS)) {
00173         ret = TRUE;
00174         goto out_free_path;
00175         }
00176 
00177         if (!osync_file_read(pidpath, &pidstr, &pidlen, error))
00178         goto out_free_path;
00179 
00180         pid = atol(pidstr);
00181         if (!pid)
00182         goto out_free_str;
00183 
00184         osync_trace(TRACE_INTERNAL, "Killing old osplugin process. PID: %ld", (long)pid);
00185 
00186         if (_osync_kill(pid, SIGTERM) < 0) {
00187         osync_trace(TRACE_INTERNAL, "Error killing old osplugin: %s. Stale pid file?", g_strerror(errno));
00188         // Don't return failure if kill() failed, because it may be a stale pid file
00189         }
00190 
00191         int count = 0;
00192         while (osync_queue_is_alive(proxy->outgoing)) {
00193         if (count++ > 10) {
00194         osync_trace(TRACE_INTERNAL, "Killing old osplugin process with SIGKILL");
00195         _osync_kill(pid, SIGKILL);
00196         break;
00197         }
00198         osync_trace(TRACE_INTERNAL, "Waiting for other side to terminate");
00199         // FIXME: Magic numbers are evil
00200         g_usleep(500000);
00201         }
00202 
00203         if (unlink(pidpath) < 0) {
00204         osync_error_set(error, OSYNC_ERROR_GENERIC, "Couldn't erase PID file: %s", g_strerror(errno));
00205         goto out_free_str;
00206         }
00207 
00208         // Success
00209         ret = TRUE;
00210 
00211         out_free_str:
00212         g_free(pidstr);
00213         out_free_path:
00214         g_free(pidpath);
00215         //out:
00216         return ret;
00217         }*/
00218 
00219 
00222 static void _osync_client_proxy_hup_handler(OSyncMessage *message, void *user_data)
00223 {
00224         OSyncClientProxy *proxy = user_data;
00225         OSyncError *error = NULL;
00226         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00227 
00228         osync_trace(TRACE_INTERNAL, "client received command %i on sending queue", osync_message_get_command(message));
00229 
00230         if ( (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR) /* Treat an error as a disconnect */
00231              || (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP) ) {
00232                 /* The remote side disconnected. So we can now disconnect as well and then
00233                  * shutdown */
00234                 if (!osync_queue_disconnect(proxy->outgoing, &error))
00235                         osync_error_unref(&error);
00236                 
00237                 if (!osync_queue_disconnect(proxy->incoming, &error))
00238                         osync_error_unref(&error);
00239                 
00240         } else {
00241                 /* This should never ever happen */
00242                 osync_trace(TRACE_ERROR, "received neither a hup, nor a error on a sending queue...");
00243         }
00244 
00245         osync_trace(TRACE_EXIT, "%s", __func__);
00246         return;
00247 }
00248 
00249 static void _osync_client_proxy_init_handler(OSyncMessage *message, void *user_data)
00250 {
00251         callContext *ctx = user_data;
00252         OSyncClientProxy *proxy = ctx->proxy;
00253         OSyncError *error = NULL;
00254         OSyncError *locerror = NULL;
00255         
00256         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00257         
00258         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00259                 ctx->init_callback(proxy, ctx->init_callback_data, NULL);
00260         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00261                 osync_demarshal_error(message, &error);
00262                 ctx->init_callback(proxy, ctx->init_callback_data, error);
00263                 osync_error_unref(&error);
00264         } else {
00265                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00266                 goto error;
00267         }
00268         
00269         osync_free(ctx);
00270         
00271         osync_trace(TRACE_EXIT, "%s", __func__);
00272         return;
00273         
00274  error:
00275         ctx->init_callback(proxy, ctx->init_callback_data, locerror);
00276         osync_free(ctx);
00277         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00278         osync_error_unref(&locerror);
00279         return;
00280 }
00281 
00282 static void _osync_client_proxy_fin_handler(OSyncMessage *message, void *user_data)
00283 {
00284         callContext *ctx = user_data;
00285         OSyncClientProxy *proxy = ctx->proxy;
00286         OSyncError *error = NULL;
00287         OSyncError *locerror = NULL;
00288         
00289         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00290         
00291         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00292                 ctx->fin_callback(proxy, ctx->fin_callback_data, NULL);
00293         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00294                 osync_demarshal_error(message, &error);
00295                 ctx->fin_callback(proxy, ctx->fin_callback_data, error);
00296                 osync_error_unref(&error);
00297         } else {
00298                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00299                 goto error;
00300         }
00301         
00302         osync_free(ctx);
00303         
00304         osync_trace(TRACE_EXIT, "%s", __func__);
00305         return;
00306         
00307  error:
00308         ctx->fin_callback(proxy, ctx->fin_callback_data, locerror);
00309         osync_free(ctx);
00310         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00311         osync_error_unref(&locerror);
00312         return;
00313 }
00314 
00315 static void _osync_client_proxy_discover_handler(OSyncMessage *message, void *user_data)
00316 {
00317         callContext *ctx = user_data;
00318         OSyncClientProxy *proxy = ctx->proxy;
00319         OSyncError *error = NULL;
00320         OSyncError *locerror = NULL;
00321         OSyncVersion *version = NULL;
00322         OSyncCapabilities *capabilities = NULL;
00323         unsigned int num_sinks = 0;
00324         unsigned int i = 0;
00325         OSyncObjTypeSink *sink = NULL;
00326         int sent_version = 0;
00327         char* str = NULL;
00328         int sent_capabilities = 0;
00329         OSyncMember *member = NULL;
00330         OSyncCapabilities *version_cap = NULL; 
00331         unsigned int num_res = 0;
00332         OSyncPluginConfig *config = NULL;
00333         OSyncPluginResource *resource = NULL;
00334         
00335         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00336         
00337         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00338                 
00339                 osync_message_read_int(message, &proxy->has_main_sink);
00340                 
00341                 osync_message_read_uint(message, &num_sinks);
00342                 
00343                 osync_trace(TRACE_INTERNAL, "main sink?: %i, num objs?: %i", proxy->has_main_sink, num_sinks);
00344                 
00345                 for (i = 0; i < num_sinks; i++) {
00346                         if (!osync_demarshal_objtype_sink(message, &sink, &locerror))
00347                                 goto error;
00348 
00349                         proxy->objtypes = g_list_append(proxy->objtypes, sink);
00350 
00351                         if (proxy->member)
00352                                 osync_member_add_objtype_sink(proxy->member, sink);
00353                 }
00354                 
00355                 /* Merger - Set the capabilities */
00356                 osync_message_read_int(message, &sent_version);
00357                 if (sent_version) {
00358                         version = osync_version_new(&locerror);
00359                         if(!version) {
00360                                 goto error;
00361                         }
00362                         
00363                         osync_message_read_string(message, &str);
00364                         osync_version_set_plugin(version, str);
00365                         osync_free(str);
00366                         osync_message_read_string(message, &str);
00367                         osync_version_set_priority(version, str);
00368                         osync_free(str);
00369                         osync_message_read_string(message, &str);
00370                         osync_version_set_vendor(version, str);
00371                         osync_free(str);
00372                         osync_message_read_string(message, &str);
00373                         osync_version_set_modelversion(version, str);
00374                         osync_free(str);
00375                         osync_message_read_string(message, &str);
00376                         osync_version_set_firmwareversion(version, str);
00377                         osync_free(str);
00378                         osync_message_read_string(message, &str);
00379                         osync_version_set_softwareversion(version, str);
00380                         osync_free(str);
00381                         osync_message_read_string(message, &str);
00382                         osync_version_set_hardwareversion(version, str);
00383                         osync_free(str);
00384                         osync_message_read_string(message, &str);
00385                         osync_version_set_identifier(version, str);
00386                         osync_free(str);        
00387                 }
00388                                 
00389                 osync_message_read_int(message, &sent_capabilities);
00390                 if (sent_capabilities) {
00391                         osync_message_read_string(message, &str);
00392                         capabilities = osync_capabilities_parse(str, strlen(str), &locerror);
00393                         osync_free(str);
00394                         if(!capabilities) {
00395                                 goto error_free_version;
00396                         }
00397                 }
00398                 
00399                 /* we set the capabilities for the member only if they are not set yet */
00400                 member = osync_client_proxy_get_member(proxy);
00401                 if (member && osync_member_get_capabilities(member) == NULL)
00402                         {
00403                                 osync_trace(TRACE_INTERNAL, "No capabilities set for the member right now. version: %p capabilities: %p\n", version, capabilities);
00404 
00405                                 /* we take our own capabilities rather then from the client */ 
00406                                 if (version)
00407                                         version_cap = osync_version_find_capabilities(version, &locerror);
00408 
00409                                 if (locerror)
00410                                         goto error_free_capabilities;
00411 
00412                                 if (version_cap) {
00413                                         if (capabilities)
00414                                                 osync_capabilities_unref(capabilities);
00415                                         capabilities = version_cap;
00416                                 }
00417 
00418                                 if (capabilities) {
00419                                         if (!osync_member_set_capabilities(member, capabilities, &locerror))
00420                                                 goto error_free_capabilities; 
00421 
00422                                         osync_capabilities_unref(capabilities);
00423                                 }
00424                         }
00425 
00426                 if (version)
00427                         osync_version_unref(version);
00428 
00429  
00430                 /* Store stuff in member configuration */
00431                 if (member) {
00432 
00433                         /* Demarshal discovered resources */
00434                         config = osync_member_get_config(member, &locerror);
00435                         if (!config)
00436                                 goto error;
00437 
00438                         osync_plugin_config_flush_resources(config);
00439                         osync_message_read_uint(message, &num_res);
00440                         for (i=0; i < num_res; i++) {
00441                                 if (!osync_demarshal_pluginresource(message, &resource, &locerror))
00442                                         goto error;
00443 
00444                                 osync_plugin_config_add_resource(config, resource);
00445                                 osync_plugin_resource_unref(resource);
00446                         }
00447 
00448                         if (!osync_member_save(member, &locerror))
00449                                 goto error;
00450                 }
00451 
00452                 ctx->discover_callback(proxy, ctx->discover_callback_data, NULL);
00453         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00454                 osync_demarshal_error(message, &error);
00455                 ctx->discover_callback(proxy, ctx->discover_callback_data, error);
00456                 osync_error_unref(&error);
00457         } else {
00458                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00459                 goto error;
00460         }
00461         
00462         osync_free(ctx);
00463         
00464         osync_trace(TRACE_EXIT, "%s", __func__);
00465         return;
00466 
00467  error_free_capabilities:
00468         if (capabilities)
00469                 osync_capabilities_unref(capabilities);
00470  error_free_version:
00471         if (version)
00472                 osync_version_unref(version);
00473  error:
00474         ctx->discover_callback(proxy, ctx->discover_callback_data, locerror);
00475         osync_free(ctx);
00476         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00477         osync_error_unref(&locerror);
00478         return;
00479 }
00480 
00481 static void _osync_client_proxy_connect_handler(OSyncMessage *message, void *user_data)
00482 {
00483         int slowsync;
00484         callContext *ctx = user_data;
00485         OSyncClientProxy *proxy = ctx->proxy;
00486         OSyncError *error = NULL;
00487         OSyncError *locerror = NULL;
00488         
00489         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00490         
00491         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00492                 osync_message_read_int(message, &slowsync);
00493                 ctx->connect_callback(proxy, ctx->connect_callback_data, slowsync, NULL);
00494         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00495                 osync_demarshal_error(message, &error);
00496                 ctx->connect_callback(proxy, ctx->connect_callback_data, FALSE, error);
00497                 osync_error_unref(&error);
00498         } else {
00499                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00500                 goto error;
00501         }
00502         
00503         osync_free(ctx);
00504         
00505         osync_trace(TRACE_EXIT, "%s", __func__);
00506         return;
00507         
00508  error:
00509         ctx->connect_callback(proxy, ctx->connect_callback_data, FALSE, locerror);
00510         osync_free(ctx);
00511         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00512         osync_error_unref(&locerror);
00513         return;
00514 }
00515 
00516 static void _osync_client_proxy_connect_done_handler(OSyncMessage *message, void *user_data)
00517 {
00518         callContext *ctx = user_data;
00519         OSyncClientProxy *proxy = ctx->proxy;
00520         OSyncError *error = NULL;
00521         OSyncError *locerror = NULL;
00522         
00523         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00524         
00525         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00526                 ctx->connect_done_callback(proxy, ctx->connect_done_callback_data, NULL);
00527         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00528                 osync_demarshal_error(message, &error);
00529                 ctx->connect_done_callback(proxy, ctx->connect_done_callback_data, error);
00530                 osync_error_unref(&error);
00531         } else {
00532                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00533                 goto error;
00534         }
00535         
00536         osync_free(ctx);
00537         
00538         osync_trace(TRACE_EXIT, "%s", __func__);
00539         return;
00540         
00541  error:
00542         ctx->connect_done_callback(proxy, ctx->connect_done_callback_data, locerror);
00543         osync_free(ctx);
00544         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00545         osync_error_unref(&locerror);
00546         return;
00547 }
00548 
00549 static void _osync_client_proxy_disconnect_handler(OSyncMessage *message, void *user_data)
00550 {
00551         callContext *ctx = user_data;
00552         OSyncClientProxy *proxy = ctx->proxy;
00553         OSyncError *error = NULL;
00554         OSyncError *locerror = NULL;
00555         
00556         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00557         
00558         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00559                 ctx->disconnect_callback(proxy, ctx->disconnect_callback_data, NULL);
00560         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00561                 osync_demarshal_error(message, &error);
00562                 ctx->disconnect_callback(proxy, ctx->disconnect_callback_data, error);
00563                 osync_error_unref(&error);
00564         } else {
00565                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00566                 goto error;
00567         }
00568         
00569         osync_free(ctx);
00570         
00571         osync_trace(TRACE_EXIT, "%s", __func__);
00572         return;
00573         
00574  error:
00575         ctx->disconnect_callback(proxy, ctx->disconnect_callback_data, locerror);
00576         osync_free(ctx);
00577         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00578         osync_error_unref(&locerror);
00579         return;
00580 }
00581 
00582 static void _osync_client_proxy_read_handler(OSyncMessage *message, void *user_data)
00583 {
00584         callContext *ctx = user_data;
00585         OSyncClientProxy *proxy = ctx->proxy;
00586         OSyncError *error = NULL;
00587         OSyncError *locerror = NULL;
00588         
00589         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00590         
00591         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00592                 ctx->read_callback(proxy, ctx->read_callback_data, NULL);
00593         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00594                 osync_demarshal_error(message, &error);
00595                 ctx->read_callback(proxy, ctx->read_callback_data, error);
00596                 osync_error_unref(&error);
00597         } else {
00598                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00599                 goto error;
00600         }
00601         
00602         osync_free(ctx);
00603         
00604         osync_trace(TRACE_EXIT, "%s", __func__);
00605         return;
00606         
00607  error:
00608         ctx->read_callback(proxy, ctx->read_callback_data, locerror);
00609         osync_free(ctx);
00610         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00611         osync_error_unref(&locerror);
00612         return;
00613 }
00614 
00615 static void _osync_client_proxy_get_changes_handler(OSyncMessage *message, void *user_data)
00616 {
00617         callContext *ctx = user_data;
00618         OSyncClientProxy *proxy = ctx->proxy;
00619         OSyncError *error = NULL;
00620         OSyncError *locerror = NULL;
00621         
00622         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00623         
00624         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00625                 ctx->get_changes_callback(proxy, ctx->get_changes_callback_data, NULL);
00626         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00627                 osync_demarshal_error(message, &error);
00628                 ctx->get_changes_callback(proxy, ctx->get_changes_callback_data, error);
00629                 osync_error_unref(&error);
00630         } else {
00631                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00632                 goto error;
00633         }
00634         
00635         osync_free(ctx);
00636         
00637         osync_trace(TRACE_EXIT, "%s", __func__);
00638         return;
00639         
00640  error:
00641         ctx->get_changes_callback(proxy, ctx->get_changes_callback_data, locerror);
00642         osync_free(ctx);
00643         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00644         osync_error_unref(&locerror);
00645         return;
00646 }
00647 
00648 static void _osync_client_proxy_commit_change_handler(OSyncMessage *message, void *user_data)
00649 {
00650         callContext *ctx = user_data;
00651         OSyncClientProxy *proxy = ctx->proxy;
00652         OSyncError *error = NULL;
00653         OSyncError *locerror = NULL;
00654         
00655         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00656         
00657         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00658                 char *uid = NULL;
00659                 osync_message_read_string(message, &uid);
00660                 ctx->commit_change_callback(proxy, ctx->commit_change_callback_data, uid, NULL);
00661                 osync_free(uid);
00662         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00663                 osync_demarshal_error(message, &error);
00664                 ctx->commit_change_callback(proxy, ctx->commit_change_callback_data, NULL, error);
00665                 osync_error_unref(&error);
00666         } else {
00667                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00668                 goto error;
00669         }
00670         
00671         osync_free(ctx);
00672         
00673         osync_trace(TRACE_EXIT, "%s", __func__);
00674         return;
00675         
00676  error:
00677         ctx->commit_change_callback(proxy, ctx->commit_change_callback_data, NULL, locerror);
00678         osync_free(ctx);
00679         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00680         osync_error_unref(&locerror);
00681         return;
00682 }
00683 
00684 static void _osync_client_proxy_committed_all_handler(OSyncMessage *message, void *user_data)
00685 {
00686         callContext *ctx = user_data;
00687         OSyncClientProxy *proxy = ctx->proxy;
00688         OSyncError *error = NULL;
00689         OSyncError *locerror = NULL;
00690         
00691         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00692         
00693         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00694                 ctx->committed_all_callback(proxy, ctx->committed_all_callback_data, NULL);
00695         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00696                 osync_demarshal_error(message, &error);
00697                 ctx->committed_all_callback(proxy, ctx->committed_all_callback_data, error);
00698                 osync_error_unref(&error);
00699         } else {
00700                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00701                 goto error;
00702         }
00703         
00704         osync_free(ctx);
00705         
00706         osync_trace(TRACE_EXIT, "%s", __func__);
00707         return;
00708         
00709  error:
00710         ctx->committed_all_callback(proxy, ctx->committed_all_callback_data, locerror);
00711         osync_free(ctx);
00712         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00713         osync_error_unref(&locerror);
00714         return;
00715 }
00716 
00717 static void _osync_client_proxy_sync_done_handler(OSyncMessage *message, void *user_data)
00718 {
00719         callContext *ctx = user_data;
00720         OSyncClientProxy *proxy = ctx->proxy;
00721         OSyncError *error = NULL;
00722         OSyncError *locerror = NULL;
00723         
00724         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00725         
00726         if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY) {
00727                 ctx->sync_done_callback(proxy, ctx->sync_done_callback_data, NULL);
00728         } else if (osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
00729                 osync_demarshal_error(message, &error);
00730                 ctx->sync_done_callback(proxy, ctx->sync_done_callback_data, error);
00731                 osync_error_unref(&error);
00732         } else {
00733                 osync_error_set(&locerror, OSYNC_ERROR_GENERIC, "Unexpected reply");
00734                 goto error;
00735         }
00736         
00737         osync_free(ctx);
00738         
00739         osync_trace(TRACE_EXIT, "%s", __func__);
00740         return;
00741         
00742  error:
00743         ctx->sync_done_callback(proxy, ctx->sync_done_callback_data, locerror);
00744         osync_free(ctx);
00745         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&locerror));
00746         osync_error_unref(&locerror);
00747         return;
00748 }
00749 
00750 static void _osync_client_proxy_message_handler(OSyncMessage *message, void *user_data)
00751 {
00752         OSyncClientProxy *proxy = user_data;
00753         OSyncError *error = NULL;
00754         OSyncChange *change = NULL;
00755 
00756         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
00757         
00758         osync_trace(TRACE_INTERNAL, "proxy received command %i", osync_message_get_command(message));
00759         switch (osync_message_get_command(message)) {
00760         case OSYNC_MESSAGE_NEW_CHANGE:
00761         case OSYNC_MESSAGE_READ_CHANGE:
00762 
00763                 osync_assert(proxy->change_callback);
00764                         
00765                 if (!osync_demarshal_change(message, &change, proxy->formatenv, &error))
00766                         goto error;
00767                         
00768                 proxy->change_callback(proxy, proxy->change_callback_data, change);
00769                         
00770                 osync_change_unref(change);
00771                 break;
00772 
00773         default:
00774                 break;
00775         }
00776         osync_trace(TRACE_EXIT, "%s", __func__);
00777         return;
00778 
00779  error:
00780         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&error));
00781         osync_error_unref(&error);
00782 }
00783 
00784 OSyncClientProxy *osync_client_proxy_new(OSyncFormatEnv *formatenv, OSyncMember *member, OSyncError **error)
00785 {
00786         OSyncClientProxy *proxy = NULL;
00787         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, formatenv, member, error);
00788 
00789         osync_assert(formatenv);
00790         
00791         proxy = osync_try_malloc0(sizeof(OSyncClientProxy), error);
00792         if (!proxy)
00793                 goto error;
00794         proxy->ref_count = 1;
00795         proxy->type = OSYNC_START_TYPE_UNKNOWN;
00796         proxy->formatenv = osync_format_env_ref(formatenv);
00797         
00798         /* TODO: Is member optional parameter? */
00799         if (member) {
00800                 proxy->member = member;
00801                 osync_member_ref(member);
00802         }
00803         
00804         osync_trace(TRACE_EXIT, "%s: %p", __func__, proxy);
00805         return proxy;
00806         
00807  error:
00808         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
00809         return NULL;
00810 }
00811 
00812 OSyncClientProxy *osync_client_proxy_ref(OSyncClientProxy *proxy)
00813 {
00814         osync_assert(proxy);
00815         
00816         g_atomic_int_inc(&(proxy->ref_count));
00817 
00818         return proxy;
00819 }
00820 
00821 void osync_client_proxy_unref(OSyncClientProxy *proxy)
00822 {
00823         OSyncObjTypeSink *sink = NULL;
00824         osync_assert(proxy);
00825         
00826         if (g_atomic_int_dec_and_test(&(proxy->ref_count))) {
00827                 if (proxy->path)
00828                         g_free(proxy->path);
00829         
00830                 if (proxy->member)
00831                         osync_member_unref(proxy->member);
00832                 
00833                 while (proxy->objtypes) {
00834                         sink = proxy->objtypes->data;
00835                         osync_objtype_sink_unref(sink);
00836                         proxy->objtypes = g_list_remove(proxy->objtypes, sink);
00837                 }
00838                 
00839                 if (proxy->context)
00840                         g_main_context_unref(proxy->context);
00841 
00842                 if (proxy->formatenv)
00843                         osync_format_env_unref(proxy->formatenv);
00844                 
00845                 osync_free(proxy);
00846         }
00847 }
00848 
00849 void osync_client_proxy_set_context(OSyncClientProxy *proxy, GMainContext *ctx)
00850 {
00851         osync_assert(proxy);
00852         proxy->context = ctx;
00853         if (ctx)
00854                 g_main_context_ref(ctx);
00855 }
00856 
00857 
00858 void osync_client_proxy_set_change_callback(OSyncClientProxy *proxy, change_cb cb, void *userdata)
00859 {
00860         osync_assert(proxy);
00861         
00862         proxy->change_callback = cb;
00863         proxy->change_callback_data = userdata;
00864 }
00865 
00866 OSyncMember *osync_client_proxy_get_member(OSyncClientProxy *proxy)
00867 {
00868         osync_assert(proxy);
00869         return proxy->member;
00870 }
00871 
00872 osync_bool osync_client_proxy_spawn(OSyncClientProxy *proxy, OSyncStartType type, const char *path, OSyncError **error)
00873 {
00874         OSyncQueue *read1 = NULL;
00875         OSyncQueue *read2 = NULL;
00876         OSyncQueue *write1 = NULL;
00877         OSyncQueue *write2 = NULL;
00878         pid_t cpid = 0;
00879         char *readfd = NULL;
00880         char *writefd = NULL;
00881         char *name = NULL;
00882         
00883         osync_trace(TRACE_ENTRY, "%s(%p, %i, %s, %p)", __func__, proxy, type, path, error);
00884         osync_assert(proxy);
00885         osync_assert(type != OSYNC_START_TYPE_UNKNOWN);
00886                 
00887         proxy->type = type;
00888         
00889         if (type != OSYNC_START_TYPE_EXTERNAL) {
00890                 // First, create the pipe from the engine to the client
00891                 if (!osync_queue_new_pipes(&read1, &write1, error))
00892                         goto error;
00893                         
00894                 // Then the pipe from the client to the engine
00895                 if (!osync_queue_new_pipes(&read2, &write2, error))
00896                         goto error_free_pipe1;
00897                 
00898                 proxy->outgoing = osync_queue_ref(write1);
00899                 proxy->incoming = osync_queue_ref(read2);
00900                 
00901                 /* Now we either spawn a new process, or we create a new thread */
00902                 if (type == OSYNC_START_TYPE_THREAD) {
00903                         proxy->client = osync_client_new(error);
00904                         if (!proxy->client)
00905                                 goto error_free_pipe2;
00906                         
00907                         /* We now connect to our incoming queue */
00908                         if (!osync_queue_connect(read1, OSYNC_QUEUE_RECEIVER, error))
00909                                 goto error_free_pipe2;
00910                         
00911                         /* and the to the outgoing queue */
00912                         if (!osync_queue_connect(write2, OSYNC_QUEUE_SENDER, error))
00913                                 goto error_free_pipe2;
00914                         
00915                         
00916                         osync_client_set_incoming_queue(proxy->client, read1);
00917                         osync_client_set_outgoing_queue(proxy->client, write2);
00918                         osync_queue_cross_link(read1, write2);
00919                         
00920                         if (!osync_client_run(proxy->client, error))
00921                                 goto error_free_pipe2;
00922                 } else {
00923                         /* First lets see if the old plugin exists, and kill it if it does */
00924                         //if (!_osync_client_kill_old_osplugin(proxy, error))
00925                         //      goto error;
00926 
00927                         //if (!osync_queue_exists(proxy->outgoing) || !osync_queue_is_alive(proxy->outgoing)) {
00928                         if (!proxy->outgoing || !osync_queue_exists(proxy->outgoing) || !osync_queue_is_alive(proxy->outgoing)) {
00929 #ifndef _WIN32
00930                                 cpid = fork();
00931                                 if (cpid == 0) {
00932                                         osync_trace_reset_indent();
00933                                         
00934                                         /* close the read and write ends of the pipes */
00935                                         osync_queue_disconnect(write1, error);
00936                                         osync_queue_disconnect(read2, error);
00937                                                 
00938                                         osync_trace(TRACE_INTERNAL, "About to exec osplugin");
00939                                         //char *memberstring = g_strdup_printf("%lli", osync_member_get_id(proxy->member));
00940                                         //execlp("osplugin", "osplugin", osync_group_get_configdir(osync_member_get_group(osync_proxy_get_member(proxy)), memberstring, NULL);
00941                                         readfd = osync_strdup_printf("%i", osync_queue_get_fd(read1));
00942                                         writefd = osync_strdup_printf("%i", osync_queue_get_fd(write2));
00943                                         execlp(OSPLUGIN, "osplugin", "-f", readfd, writefd, NULL);
00944 
00945                                         if (errno == ENOENT) {
00946                                                 osync_trace(TRACE_INTERNAL, "Unable to find osplugin. Trying local path.");
00947                                                 //execlp("osplugin", "osplugin", osync_group_get_configdir(osync_member_get_group(osync_proxy_get_member(proxy)), memberstring, NULL);
00948                                                 execlp("./osplugin", "osplugin", "-f", readfd, writefd, NULL);
00949                                         }
00950                                                                                         
00951                                         osync_trace(TRACE_INTERNAL, strerror(errno));
00952                                         osync_trace(TRACE_INTERNAL, "Unable to execute osplugin.");
00953                                         exit(1);
00954                                 } else {
00955                                         /* close the read and write ends of the pipes */
00956                                         osync_queue_disconnect(write2, error);
00957                                         osync_queue_disconnect(read1, error);
00958                                 }
00959         
00960                                 proxy->child_pid = cpid;
00961                         
00962                                 //while (!osync_queue_exists(proxy->outgoing)) {
00963                                 //      osync_trace(TRACE_INTERNAL, "Waiting for other side to create fifo");
00964                                 //      g_usleep(500000);
00965                                 //}
00966                         
00967                                 osync_trace(TRACE_INTERNAL, "Queue was created");
00968 #endif //_WIN32
00969                         }
00970         
00971                         //if (proxy->child_pid) {
00972                         //      if (!_osync_client_create_pidfile(proxy, error))
00973                         //              goto error;
00974                         //}
00975                 }
00976                 
00977                 osync_queue_unref(read1);
00978                 osync_queue_unref(write1);
00979                 osync_queue_unref(read2);
00980                 osync_queue_unref(write2);
00981 
00982                 /* We now connect to our incoming queue */
00983                 if (!osync_queue_connect(proxy->incoming, OSYNC_QUEUE_RECEIVER, error))
00984                         goto error;
00985                         
00986                 /* and the to the outgoing queue */
00987                 if (!osync_queue_connect(proxy->outgoing, OSYNC_QUEUE_SENDER, error))
00988                         goto error;
00989         } else {
00990                 name = osync_strdup_printf("%s%cpluginpipe", path, G_DIR_SEPARATOR);
00991                 proxy->outgoing = osync_queue_new(name, error);
00992                 osync_free(name);
00993                 if (!proxy->outgoing)
00994                         goto error;
00995                 
00996                 name = osync_strdup_printf("%s%cenginepipe", path, G_DIR_SEPARATOR);
00997                 proxy->incoming = osync_queue_new(name, error);
00998                 osync_free(name);
00999                 if (!proxy->incoming)
01000                         goto error;
01001                         
01002                 if (!osync_queue_create(proxy->outgoing, error))
01003                         goto error;
01004                         
01005                 if (!osync_queue_create(proxy->incoming, error))
01006                         goto error;
01007                         
01008                 /* and the to the outgoing queue */
01009                 if (!osync_queue_connect(proxy->outgoing, OSYNC_QUEUE_SENDER, error))
01010                         goto error;
01011         }
01012         
01013         osync_queue_set_message_handler(proxy->incoming, _osync_client_proxy_message_handler, proxy);
01014         osync_queue_setup_with_gmainloop(proxy->incoming, proxy->context);
01015         
01016         osync_queue_set_message_handler(proxy->outgoing, _osync_client_proxy_hup_handler, proxy);
01017         osync_queue_setup_with_gmainloop(proxy->outgoing, proxy->context);
01018         
01019         osync_trace(TRACE_EXIT, "%s", __func__);
01020         return TRUE;
01021         
01022  error_free_pipe2:
01023         osync_queue_unref(read2);
01024         osync_queue_unref(write2);
01025  error_free_pipe1:
01026         osync_queue_unref(read1);
01027         osync_queue_unref(write1);
01028  error:
01029         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01030         return FALSE;
01031 }
01032 
01033 osync_bool osync_client_proxy_shutdown(OSyncClientProxy *proxy, OSyncError **error)
01034 {
01035         OSyncMessage *message = NULL;
01036         int status = 0;
01037         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, proxy, error);
01038         
01039         /* We first disconnect our reading queue. This will generate a HUP
01040          * on the remote side */
01041         if (!osync_queue_disconnect(proxy->incoming, error))
01042                 goto error;
01043         
01044         /* We now wait for the HUP on our sending queue */
01045         message = osync_queue_get_message(proxy->outgoing);
01046         if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
01047                 osync_error_set(error, OSYNC_ERROR_GENERIC, "Disconnected, but received no HUP");
01048                 osync_message_unref(message);
01049                 goto error;
01050         }
01051         osync_message_unref(message);
01052         
01053         /* After we received the HUP, we can disconnect */
01054         if (!osync_queue_disconnect(proxy->outgoing, error))
01055                 goto error;
01056                         
01057         if (proxy->type == OSYNC_START_TYPE_THREAD) {
01058 #ifndef OPENSYNC_PREVENT_CLIENT_SHUTDOWN
01059                 osync_client_shutdown(proxy->client);
01060 #endif
01061                 osync_client_unref(proxy->client);
01062         } else if (proxy->type == OSYNC_START_TYPE_PROCESS) {
01063                 if (proxy->child_pid) {
01064 #ifndef _WIN32
01065                         if (waitpid(proxy->child_pid, &status, 0) == -1) {
01066                                 osync_error_set(error, OSYNC_ERROR_GENERIC, "Error waiting for osplugin process: %s", g_strerror(errno));
01067                                 goto error;
01068                         }
01069 
01070                         if (!WIFEXITED(status))
01071                                 osync_trace(TRACE_INTERNAL, "Child has exited abnormally");
01072                         else if (WEXITSTATUS(status) != 0)
01073                                 osync_trace(TRACE_INTERNAL, "Child has returned non-zero exit status (%d)", WEXITSTATUS(status));
01074 #endif //_WIN32
01075 
01076                         //if (!osync_client_remove_pidfile(client, error))
01077                         //      goto error;
01078                 }
01079                 
01080                 /* First lets see if the old plugin exists, and kill it if it does */
01081                 //if (!_osync_client_kill_old_osplugin(proxy, error))
01082                 //      goto error;
01083         }
01084                         
01085         osync_queue_unref(proxy->incoming);
01086         osync_queue_unref(proxy->outgoing);
01087         
01088         osync_trace(TRACE_EXIT, "%s", __func__);
01089         return TRUE;
01090         
01091  error:
01092         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01093         return FALSE;
01094 }
01095 
01108 static osync_bool osync_client_proxy_check_resource(OSyncClientProxy *proxy, OSyncPluginResource *resource, OSyncError **error)
01109 {
01110         OSyncList *format_sinks = osync_plugin_resource_get_objformat_sinks(resource);
01111 
01112         /* OSyncPluginConfig should fail on validiation if format sinks are missing for any resource */
01113         osync_assert(format_sinks);
01114 
01115         for (; format_sinks; format_sinks = format_sinks->next) {
01116                 OSyncObjFormatSink *format_sink = format_sinks->data;   
01117                 const char *objformat_name = osync_objformat_sink_get_objformat(format_sink);
01118 
01119                 if (!osync_format_env_find_objformat(proxy->formatenv, objformat_name)) {
01120                         osync_error_set(error, OSYNC_ERROR_PLUGIN_NOT_FOUND, "Plugin for format \"%s\" not found.", objformat_name);
01121                         return FALSE;
01122                 }
01123         }
01124 
01125         return TRUE;
01126 }
01127 
01128 osync_bool osync_client_proxy_initialize(OSyncClientProxy *proxy, initialize_cb callback, void *userdata, const char *formatdir, const char *plugindir, const char *plugin, const char *groupname, const char *configdir, OSyncPluginConfig *config, OSyncError **error)
01129 {
01130         callContext *ctx = NULL;
01131         int haspluginconfig = config ? TRUE : FALSE;
01132         OSyncMessage *message = NULL;
01133 #ifdef OPENSYNC_UNITTESTS
01134         long long int memberid = 0;
01135 #endif
01136         
01137         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %s, %s, %s, %p, %p)", __func__, proxy, callback, userdata, formatdir, plugindir, plugin, groupname, configdir, config, error);
01138         osync_assert(proxy);
01139         
01140 
01141         ctx = osync_try_malloc0(sizeof(callContext), error);
01142         if (!ctx)
01143                 goto error;
01144         
01145         ctx->proxy = proxy;
01146         ctx->init_callback = callback;
01147         ctx->init_callback_data = userdata;
01148         
01149         message = osync_message_new(OSYNC_MESSAGE_INITIALIZE, 0, error);
01150         if (!message)
01151                 goto error;
01152         
01153         osync_message_write_string(message, osync_queue_get_path(proxy->incoming));
01154         osync_message_write_string(message, formatdir);
01155         osync_message_write_string(message, plugindir);
01156         osync_message_write_string(message, plugin);
01157         osync_message_write_string(message, groupname);
01158         osync_message_write_string(message, configdir);
01159         osync_message_write_int(message, haspluginconfig);
01160 
01161         if (haspluginconfig && !osync_marshal_pluginconfig(message, config, error))
01162                 goto error;
01163 
01164         if (haspluginconfig) {
01165                 OSyncList *r = osync_plugin_config_get_resources(config);
01166                 for (; r; r = r->next) {
01167                         OSyncPluginResource *res = r->data;
01168                         const char *objtype = NULL;
01169                         OSyncObjTypeSink *sink = NULL;
01170 
01171 
01172                         if (!osync_plugin_resource_is_enabled(res))
01173                                 continue;
01174 
01175                         if (!osync_client_proxy_check_resource(proxy, res, error))
01176                                 goto error;
01177 
01178                         objtype = osync_plugin_resource_get_objtype(res);
01179                         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01180                         /* TODO: In discovery phase *sink COULD be NULL. Review if this is correct behavior. */
01181                         if (sink) {
01182                                 osync_objtype_sink_ref(sink);
01183                                 proxy->objtypes = g_list_append(proxy->objtypes, sink);
01184                         }
01185                 }
01186         }
01187 
01188 #ifdef OPENSYNC_UNITTESTS
01189         // Introduced (only) for testing/debugging purpose (mock-sync)
01190 
01191         if (proxy->member)
01192                 memberid = osync_member_get_id(proxy->member);
01193 
01194         osync_message_write_long_long_int(message, memberid);
01195 #endif  
01196         
01197         osync_message_set_handler(message, _osync_client_proxy_init_handler, ctx);
01198         
01199         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, proxy->timeout.initialize, error))
01200                 goto error_free_message;
01201         
01202         osync_message_unref(message);
01203         
01204         if (proxy->type == OSYNC_START_TYPE_EXTERNAL) {
01205                 /* We now connect to our incoming queue */
01206                 if (!osync_queue_connect(proxy->incoming, OSYNC_QUEUE_RECEIVER, error))
01207                         goto error;
01208         }
01209         
01210         osync_trace(TRACE_EXIT, "%s", __func__);
01211         return TRUE;
01212 
01213  error_free_message:
01214         osync_message_unref(message);
01215  error:
01216         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01217         return FALSE;
01218 }
01219 
01220 unsigned int osync_client_proxy_get_initialize_timeout(OSyncClientProxy *proxy)
01221 {
01222         osync_assert(proxy);
01223         return proxy->timeout.initialize;
01224 }
01225 
01226 void osync_client_proxy_set_initialize_timeout(OSyncClientProxy *proxy, unsigned int timeout)
01227 {
01228         osync_assert(proxy);
01229         proxy->timeout.initialize = timeout;
01230 }
01231 
01232 osync_bool osync_client_proxy_finalize(OSyncClientProxy *proxy, finalize_cb callback, void *userdata, OSyncError **error)
01233 {
01234         callContext *ctx = NULL;
01235         OSyncMessage *message = NULL;
01236 
01237         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %p)", __func__, proxy, callback, userdata, error);
01238         
01239         ctx = osync_try_malloc0(sizeof(callContext), error);
01240         if (!ctx)
01241                 goto error;
01242         
01243         ctx->proxy = proxy;
01244         ctx->fin_callback = callback;
01245         ctx->fin_callback_data = userdata;
01246         
01247         message = osync_message_new(OSYNC_MESSAGE_FINALIZE, 0, error);
01248         if (!message)
01249                 goto error;
01250 
01251         osync_message_set_handler(message, _osync_client_proxy_fin_handler, ctx);
01252         
01253         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, proxy->timeout.finalize, error))
01254                 goto error_free_message;
01255 
01256         osync_message_unref(message);
01257 
01258         osync_trace(TRACE_EXIT, "%s", __func__);
01259         return TRUE;
01260 
01261  error_free_message:
01262         osync_message_unref(message);
01263  error:
01264         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01265         return FALSE;
01266 }
01267 
01268 unsigned int osync_client_proxy_get_finalize_timeout(OSyncClientProxy *proxy)
01269 {
01270         osync_assert(proxy);
01271         return proxy->timeout.finalize;
01272 }
01273 
01274 void osync_client_proxy_set_finalize_timeout(OSyncClientProxy *proxy, unsigned int timeout)
01275 {
01276         osync_assert(proxy);
01277         proxy->timeout.finalize = timeout;
01278 }
01279 
01280 osync_bool osync_client_proxy_discover(OSyncClientProxy *proxy, discover_cb callback, void *userdata, OSyncError **error)
01281 {
01282         callContext *ctx = NULL;
01283         OSyncMessage *message = NULL;
01284 
01285         osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, proxy, callback, userdata, error);
01286         
01287         ctx = osync_try_malloc0(sizeof(callContext), error);
01288         if (!ctx)
01289                 goto error;
01290         
01291         ctx->proxy = proxy;
01292         ctx->discover_callback = callback;
01293         ctx->discover_callback_data = userdata;
01294         
01295         message = osync_message_new(OSYNC_MESSAGE_DISCOVER, 0, error);
01296         if (!message)
01297                 goto error;
01298 
01299         osync_message_set_handler(message, _osync_client_proxy_discover_handler, ctx);
01300         
01301         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, proxy->timeout.discover, error))
01302                 goto error_free_message;
01303         
01304         osync_message_unref(message);
01305 
01306         osync_trace(TRACE_EXIT, "%s", __func__);
01307         return TRUE;
01308 
01309  error_free_message:
01310         osync_message_unref(message);
01311  error:
01312         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01313         return FALSE;
01314 }
01315 
01316 unsigned int osync_client_proxy_get_discover_timeout(OSyncClientProxy *proxy)
01317 {
01318         osync_assert(proxy);
01319         return proxy->timeout.discover;
01320 }
01321 
01322 void osync_client_proxy_set_discover_timeout(OSyncClientProxy *proxy, unsigned int timeout)
01323 {
01324         osync_assert(proxy);
01325         proxy->timeout.discover = timeout;
01326 }
01327 
01328 int osync_client_proxy_num_objtypes(OSyncClientProxy *proxy)
01329 {
01330         osync_assert(proxy);
01331         return g_list_length(proxy->objtypes);
01332 }
01333 
01334 OSyncObjTypeSink *osync_client_proxy_nth_objtype(OSyncClientProxy *proxy, int nth)
01335 {
01336         osync_assert(proxy);
01337         return g_list_nth_data(proxy->objtypes, nth);
01338 }
01339 
01340 OSyncObjTypeSink *osync_client_proxy_find_objtype_sink(OSyncClientProxy *proxy, const char *objtype)
01341 {
01342         GList *o = NULL;
01343         OSyncObjTypeSink *sink = NULL;
01344 
01345         osync_assert(proxy);
01346 
01347         for (o = proxy->objtypes; o; o = o->next) {
01348                 sink = o->data;
01349                 if (!objtype && !osync_objtype_sink_get_name(sink))
01350                         return sink;
01351 
01352                 if (objtype && !strcmp(osync_objtype_sink_get_name(sink), objtype))
01353                         return sink;
01354         }
01355 
01356         if (objtype && proxy->member)
01357                 return osync_member_find_objtype_sink(proxy->member, objtype);
01358         else if (!objtype && proxy->member)
01359                 return osync_member_get_main_sink(proxy->member);
01360 
01361         return NULL;
01362 }
01363 
01364 osync_bool osync_client_proxy_connect(OSyncClientProxy *proxy, connect_cb callback, void *userdata, const char *objtype, osync_bool slowsync, OSyncError **error)
01365 {
01366         int timeout = 0;
01367         callContext *ctx = NULL;
01368         OSyncObjTypeSink *sink = NULL;
01369         OSyncMessage *message = NULL;
01370         
01371         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %i, %p)", __func__, proxy, callback, userdata, objtype, slowsync, error);
01372         
01373         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_CONNECT;
01374 
01375         ctx = osync_try_malloc0(sizeof(callContext), error);
01376         if (!ctx)
01377                 goto error;
01378         
01379         ctx->proxy = proxy;
01380         ctx->connect_callback = callback;
01381         ctx->connect_callback_data = userdata;
01382         
01383         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01384         if (sink)
01385                 timeout = osync_objtype_sink_get_connect_timeout_or_default(sink); 
01386 
01387         message = osync_message_new(OSYNC_MESSAGE_CONNECT, 0, error);
01388         if (!message)
01389                 goto error_free_context;
01390         
01391         osync_message_set_handler(message, _osync_client_proxy_connect_handler, ctx);
01392 
01393         osync_message_write_string(message, objtype);
01394         osync_message_write_int(message, slowsync);
01395 
01396         
01397         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01398                 goto error_free_message;
01399         
01400         osync_message_unref(message);
01401         
01402         osync_trace(TRACE_EXIT, "%s", __func__);
01403         return TRUE;
01404 
01405  error_free_message:
01406         osync_message_unref(message);
01407  error_free_context:
01408         osync_free(ctx);
01409  error:
01410         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01411         return FALSE;
01412 }
01413 
01414 osync_bool osync_client_proxy_connect_done(OSyncClientProxy *proxy, sync_done_cb callback, void *userdata, const char *objtype, OSyncError **error)
01415 {
01416         int timeout = 0;
01417         callContext *ctx = NULL;
01418         OSyncObjTypeSink *sink = NULL;
01419         OSyncMessage *message = NULL;
01420 
01421         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %p)", __func__, proxy, callback, userdata, objtype, error);
01422         osync_assert(proxy);
01423 
01424         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_CONNECTDONE;
01425 
01426         ctx = osync_try_malloc0(sizeof(callContext), error);
01427         if (!ctx)
01428                 goto error;
01429 
01430         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01431         if (sink)
01432                 timeout = osync_objtype_sink_get_connectdone_timeout_or_default(sink); 
01433 
01434         ctx->proxy = proxy;
01435         ctx->connect_done_callback = callback;
01436         ctx->connect_done_callback_data = userdata;
01437         
01438         message = osync_message_new(OSYNC_MESSAGE_CONNECT_DONE, 0, error);
01439         if (!message)
01440                 goto error_free_context;
01441         
01442         osync_message_set_handler(message, _osync_client_proxy_connect_done_handler, ctx);
01443 
01444         osync_message_write_string(message, objtype);
01445         
01446         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01447                 goto error_free_message;
01448         
01449         osync_message_unref(message);
01450         
01451         osync_trace(TRACE_EXIT, "%s", __func__);
01452         return TRUE;
01453 
01454  error_free_message:
01455         osync_message_unref(message);
01456  error_free_context:
01457         osync_free(ctx);
01458  error:
01459         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01460         return FALSE;
01461 }
01462 
01463 osync_bool osync_client_proxy_disconnect(OSyncClientProxy *proxy, disconnect_cb callback, void *userdata, const char *objtype, OSyncError **error)
01464 {
01465         int timeout = 0;
01466         callContext *ctx = NULL;
01467         OSyncObjTypeSink *sink = NULL;
01468         OSyncMessage *message = NULL;
01469 
01470         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %p)", __func__, proxy, callback, userdata, objtype, error);
01471 
01472         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_DISCONNECT;
01473 
01474         ctx = osync_try_malloc0(sizeof(callContext), error);
01475         if (!ctx)
01476                 goto error;
01477         
01478         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01479         if (sink)
01480                 timeout = osync_objtype_sink_get_disconnect_timeout_or_default(sink); 
01481 
01482         ctx->proxy = proxy;
01483         ctx->disconnect_callback = callback;
01484         ctx->disconnect_callback_data = userdata;
01485         
01486         message = osync_message_new(OSYNC_MESSAGE_DISCONNECT, 0, error);
01487         if (!message)
01488                 goto error_free_context;
01489         
01490         osync_message_set_handler(message, _osync_client_proxy_disconnect_handler, ctx);
01491 
01492         osync_message_write_string(message, objtype);
01493         
01494         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01495                 goto error_free_message;
01496         
01497         osync_message_unref(message);
01498         
01499         osync_trace(TRACE_EXIT, "%s", __func__);
01500         return TRUE;
01501 
01502  error_free_message:
01503         osync_message_unref(message);
01504  error_free_context:
01505         osync_free(ctx);
01506  error:
01507         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01508         return FALSE;
01509 }
01510 
01511 osync_bool osync_client_proxy_read(OSyncClientProxy *proxy, read_cb callback, void *userdata, OSyncChange *change, OSyncError **error)
01512 {
01513         int timeout = 0;
01514         callContext *ctx = NULL;
01515         OSyncObjTypeSink *sink = NULL;
01516         OSyncMessage *message = NULL;
01517 
01518         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %p, %p)", __func__, proxy, callback, userdata, change, error);
01519         
01520         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_READ;
01521 
01522         ctx = osync_try_malloc0(sizeof(callContext), error);
01523         if (!ctx)
01524                 goto error;
01525         
01526         sink = osync_client_proxy_find_objtype_sink(proxy, osync_change_get_objtype(change));
01527         if (sink)
01528                 timeout = osync_objtype_sink_get_read_timeout_or_default(sink); 
01529 
01530         ctx->proxy = proxy;
01531         ctx->read_callback = callback;
01532         ctx->read_callback_data = userdata;
01533         
01534         message = osync_message_new(OSYNC_MESSAGE_READ_CHANGE, 0, error);
01535         if (!message)
01536                 goto error_free_context;
01537         
01538         osync_message_set_handler(message, _osync_client_proxy_read_handler, ctx);
01539 
01540         if (!osync_marshal_change(message, change, error))
01541                 goto error_free_message;
01542 
01543         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01544                 goto error_free_message;
01545 
01546         osync_message_unref(message);
01547         
01548         osync_trace(TRACE_EXIT, "%s", __func__);
01549         return TRUE;
01550 
01551  error_free_message:
01552         osync_message_unref(message);
01553  error_free_context:
01554         osync_free(ctx);
01555  error:
01556         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01557         return FALSE;
01558 }
01559 
01560 osync_bool osync_client_proxy_get_changes(OSyncClientProxy *proxy, get_changes_cb callback, void *userdata, const char *objtype, osync_bool slowsync, OSyncError **error)
01561 {
01562         int timeout = 0;
01563         callContext *ctx = NULL;
01564         OSyncObjTypeSink *sink = NULL;
01565         OSyncMessage *message = NULL;
01566 
01567         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %i, %p)", __func__, proxy, callback, userdata, objtype, slowsync, error);
01568         
01569         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_GETCHANGES;
01570 
01571         ctx = osync_try_malloc0(sizeof(callContext), error);
01572         if (!ctx)
01573                 goto error;
01574         
01575         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01576         if (sink)
01577                 timeout = osync_objtype_sink_get_getchanges_timeout_or_default(sink); 
01578 
01579         ctx->proxy = proxy;
01580         ctx->get_changes_callback = callback;
01581         ctx->get_changes_callback_data = userdata;
01582         
01583         message = osync_message_new(OSYNC_MESSAGE_GET_CHANGES, 0, error);
01584         if (!message)
01585                 goto error_free_context;
01586         
01587         osync_message_set_handler(message, _osync_client_proxy_get_changes_handler, ctx);
01588 
01589         osync_message_write_string(message, objtype);
01590         osync_message_write_int(message, slowsync);
01591         
01592         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01593                 goto error_free_message;
01594         
01595         osync_message_unref(message);
01596         
01597         osync_trace(TRACE_EXIT, "%s", __func__);
01598         return TRUE;
01599 
01600  error_free_message:
01601         osync_message_unref(message);
01602  error_free_context:
01603         osync_free(ctx);
01604  error:
01605         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01606         return FALSE;
01607 }
01608 
01609 osync_bool osync_client_proxy_commit_change(OSyncClientProxy *proxy, commit_change_cb callback, void *userdata, OSyncChange *change, OSyncError **error)
01610 {
01611         int timeout = 0;
01612         callContext *ctx = NULL;
01613         OSyncObjTypeSink *sink = NULL;
01614         OSyncMessage *message = NULL;
01615 
01616         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %p, %p)", __func__, proxy, callback, userdata, change, error);
01617         osync_assert(proxy);
01618         osync_assert(change);
01619 
01620         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_COMMIT;
01621 
01622         ctx = osync_try_malloc0(sizeof(callContext), error);
01623         if (!ctx)
01624                 goto error;
01625 
01626         sink = osync_client_proxy_find_objtype_sink(proxy, osync_change_get_objtype(change));
01627         if (sink)
01628                 timeout = osync_objtype_sink_get_commit_timeout_or_default(sink); 
01629         
01630         ctx->proxy = proxy;
01631         ctx->commit_change_callback = callback;
01632         ctx->commit_change_callback_data = userdata;
01633         
01634         message = osync_message_new(OSYNC_MESSAGE_COMMIT_CHANGE, 0, error);
01635         if (!message)
01636                 goto error_free_context;
01637         
01638         osync_message_set_handler(message, _osync_client_proxy_commit_change_handler, ctx);
01639 
01640         if (!osync_marshal_change(message, change, error))
01641                 goto error_free_message;
01642         
01643         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01644                 goto error_free_message;
01645         
01646         osync_message_unref(message);
01647         
01648         osync_trace(TRACE_EXIT, "%s", __func__);
01649         return TRUE;
01650 
01651  error_free_message:
01652         osync_message_unref(message);
01653  error_free_context:
01654         osync_free(ctx);
01655  error:
01656         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01657         return FALSE;
01658 }
01659 
01660 osync_bool osync_client_proxy_committed_all(OSyncClientProxy *proxy, committed_all_cb callback, void *userdata, const char *objtype, OSyncError **error)
01661 {
01662         int timeout = 0;
01663         callContext *ctx = NULL;
01664         OSyncObjTypeSink *sink = NULL;
01665         OSyncMessage *message = NULL;
01666 
01667         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %p)", __func__, proxy, callback, userdata, objtype, error);
01668         osync_assert(proxy);
01669 
01670         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_COMMITTEDALL;
01671         
01672         ctx = osync_try_malloc0(sizeof(callContext), error);
01673         if (!ctx)
01674                 goto error;
01675         
01676         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01677         if (sink)
01678                 timeout = osync_objtype_sink_get_committedall_timeout_or_default(sink); 
01679 
01680         ctx->proxy = proxy;
01681         ctx->committed_all_callback = callback;
01682         ctx->committed_all_callback_data = userdata;
01683         
01684         message = osync_message_new(OSYNC_MESSAGE_COMMITTED_ALL, 0, error);
01685         if (!message)
01686                 goto error_free_context;
01687         
01688         osync_message_set_handler(message, _osync_client_proxy_committed_all_handler, ctx);
01689 
01690         osync_message_write_string(message, objtype);
01691         
01692         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01693                 goto error_free_message;
01694         
01695         osync_message_unref(message);
01696         
01697         osync_trace(TRACE_EXIT, "%s", __func__);
01698         return TRUE;
01699 
01700  error_free_message:
01701         osync_message_unref(message);
01702  error_free_context:
01703         osync_free(ctx);
01704  error:
01705         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01706         return FALSE;
01707 }
01708 
01709 osync_bool osync_client_proxy_sync_done(OSyncClientProxy *proxy, sync_done_cb callback, void *userdata, const char *objtype, OSyncError **error)
01710 {
01711         int timeout = 0;
01712         callContext *ctx = NULL;
01713         OSyncObjTypeSink *sink = NULL;
01714         OSyncMessage *message = NULL;
01715 
01716         osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %s, %p)", __func__, proxy, callback, userdata, objtype, error);
01717         osync_assert(proxy);
01718 
01719         timeout = OSYNC_CLIENT_PROXY_TIMEOUT_SYNCDONE;
01720 
01721         ctx = osync_try_malloc0(sizeof(callContext), error);
01722         if (!ctx)
01723                 goto error;
01724 
01725         sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
01726         if (sink)
01727                 timeout = osync_objtype_sink_get_syncdone_timeout_or_default(sink); 
01728 
01729         ctx->proxy = proxy;
01730         ctx->sync_done_callback = callback;
01731         ctx->sync_done_callback_data = userdata;
01732         
01733         message = osync_message_new(OSYNC_MESSAGE_SYNC_DONE, 0, error);
01734         if (!message)
01735                 goto error_free_context;
01736         
01737         osync_message_set_handler(message, _osync_client_proxy_sync_done_handler, ctx);
01738 
01739         osync_message_write_string(message, objtype);
01740         
01741         if (!osync_queue_send_message_with_timeout(proxy->outgoing, proxy->incoming, message, timeout, error))
01742                 goto error_free_message;
01743         
01744         osync_message_unref(message);
01745         
01746         osync_trace(TRACE_EXIT, "%s", __func__);
01747         return TRUE;
01748 
01749  error_free_message:
01750         osync_message_unref(message);
01751  error_free_context:
01752         osync_free(ctx);
01753  error:
01754         osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(error));
01755         return FALSE;
01756 }
/var/data/www/opensync/code/opensync/client/opensync_client_proxy.c
