You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/08/23 14:11:58 UTC
[06/20] zookeeper git commit: ZOOKEEPER-3031: 3.4: MAVEN MIGRATION -
move client dir
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/recordio.c
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/recordio.c b/zookeeper-client/zookeeper-client-c/src/recordio.c
new file mode 100644
index 0000000..0fcf48e
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/recordio.c
@@ -0,0 +1,366 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <recordio.h>
+#include <string.h>
+#include <stdio.h>
+#include <errno.h>
+#include <stdlib.h>
+#ifndef WIN32
+#include <netinet/in.h>
+#else
+#include <winsock2.h> /* for _htonl and _ntohl */
+#endif
+
+void deallocate_String(char **s)
+{
+ if (*s)
+ free(*s);
+ *s = 0;
+}
+
+void deallocate_Buffer(struct buffer *b)
+{
+ if (b->buff)
+ free(b->buff);
+ b->buff = 0;
+}
+
+struct buff_struct {
+ int32_t len;
+ int32_t off;
+ char *buffer;
+};
+
+static int resize_buffer(struct buff_struct *s, int newlen)
+{
+ char *buffer= NULL;
+ while (s->len < newlen) {
+ s->len *= 2;
+ }
+ buffer = (char*)realloc(s->buffer, s->len);
+ if (!buffer) {
+ s->buffer = 0;
+ return -ENOMEM;
+ }
+ s->buffer = buffer;
+ return 0;
+}
+
+int oa_start_record(struct oarchive *oa, const char *tag)
+{
+ return 0;
+}
+int oa_end_record(struct oarchive *oa, const char *tag)
+{
+ return 0;
+}
+int oa_serialize_int(struct oarchive *oa, const char *tag, const int32_t *d)
+{
+ struct buff_struct *priv = oa->priv;
+ int32_t i = htonl(*d);
+ if ((priv->len - priv->off) < sizeof(i)) {
+ int rc = resize_buffer(priv, priv->len + sizeof(i));
+ if (rc < 0) return rc;
+ }
+ memcpy(priv->buffer+priv->off, &i, sizeof(i));
+ priv->off+=sizeof(i);
+ return 0;
+}
+int64_t zoo_htonll(int64_t v)
+{
+ int i = 0;
+ char *s = (char *)&v;
+ if (htonl(1) == 1) {
+ return v;
+ }
+ for (i = 0; i < 4; i++) {
+ int tmp = s[i];
+ s[i] = s[8-i-1];
+ s[8-i-1] = tmp;
+ }
+
+ return v;
+}
+
+int oa_serialize_long(struct oarchive *oa, const char *tag, const int64_t *d)
+{
+ const int64_t i = zoo_htonll(*d);
+ struct buff_struct *priv = oa->priv;
+ if ((priv->len - priv->off) < sizeof(i)) {
+ int rc = resize_buffer(priv, priv->len + sizeof(i));
+ if (rc < 0) return rc;
+ }
+ memcpy(priv->buffer+priv->off, &i, sizeof(i));
+ priv->off+=sizeof(i);
+ return 0;
+}
+int oa_start_vector(struct oarchive *oa, const char *tag, const int32_t *count)
+{
+ return oa_serialize_int(oa, tag, count);
+}
+int oa_end_vector(struct oarchive *oa, const char *tag)
+{
+ return 0;
+}
+int oa_serialize_bool(struct oarchive *oa, const char *name, const int32_t *i)
+{
+ //return oa_serialize_int(oa, name, i);
+ struct buff_struct *priv = oa->priv;
+ if ((priv->len - priv->off) < 1) {
+ int rc = resize_buffer(priv, priv->len + 1);
+ if (rc < 0)
+ return rc;
+ }
+ priv->buffer[priv->off] = (*i == 0 ? '\0' : '\1');
+ priv->off++;
+ return 0;
+}
+static const int32_t negone = -1;
+int oa_serialize_buffer(struct oarchive *oa, const char *name,
+ const struct buffer *b)
+{
+ struct buff_struct *priv = oa->priv;
+ int rc;
+ if (!b) {
+ return oa_serialize_int(oa, "len", &negone);
+ }
+ rc = oa_serialize_int(oa, "len", &b->len);
+ if (rc < 0)
+ return rc;
+ // this means a buffer of NUll
+ // with size of -1. This is
+ // waht we use in java serialization for NULL
+ if (b->len == -1) {
+ return rc;
+ }
+ if ((priv->len - priv->off) < b->len) {
+ rc = resize_buffer(priv, priv->len + b->len);
+ if (rc < 0)
+ return rc;
+ }
+ memcpy(priv->buffer+priv->off, b->buff, b->len);
+ priv->off += b->len;
+ return 0;
+}
+int oa_serialize_string(struct oarchive *oa, const char *name, char **s)
+{
+ struct buff_struct *priv = oa->priv;
+ int32_t len;
+ int rc;
+ if (!*s) {
+ oa_serialize_int(oa, "len", &negone);
+ return 0;
+ }
+ len = strlen(*s);
+ rc = oa_serialize_int(oa, "len", &len);
+ if (rc < 0)
+ return rc;
+ if ((priv->len - priv->off) < len) {
+ rc = resize_buffer(priv, priv->len + len);
+ if (rc < 0)
+ return rc;
+ }
+ memcpy(priv->buffer+priv->off, *s, len);
+ priv->off += len;
+ return 0;
+}
+int ia_start_record(struct iarchive *ia, const char *tag)
+{
+ return 0;
+}
+int ia_end_record(struct iarchive *ia, const char *tag)
+{
+ return 0;
+}
+int ia_deserialize_int(struct iarchive *ia, const char *tag, int32_t *count)
+{
+ struct buff_struct *priv = ia->priv;
+ if ((priv->len - priv->off) < sizeof(*count)) {
+ return -E2BIG;
+ }
+ memcpy(count, priv->buffer+priv->off, sizeof(*count));
+ priv->off+=sizeof(*count);
+ *count = ntohl(*count);
+ return 0;
+}
+
+int ia_deserialize_long(struct iarchive *ia, const char *tag, int64_t *count)
+{
+ struct buff_struct *priv = ia->priv;
+ int64_t v = 0;
+ if ((priv->len - priv->off) < sizeof(*count)) {
+ return -E2BIG;
+ }
+ memcpy(count, priv->buffer+priv->off, sizeof(*count));
+ priv->off+=sizeof(*count);
+ v = zoo_htonll(*count); // htonll and ntohll do the same
+ *count = v;
+ return 0;
+}
+int ia_start_vector(struct iarchive *ia, const char *tag, int32_t *count)
+{
+ return ia_deserialize_int(ia, tag, count);
+}
+int ia_end_vector(struct iarchive *ia, const char *tag)
+{
+ return 0;
+}
+int ia_deserialize_bool(struct iarchive *ia, const char *name, int32_t *v)
+{
+ struct buff_struct *priv = ia->priv;
+ //fprintf(stderr, "Deserializing bool %d\n", priv->off);
+ //return ia_deserialize_int(ia, name, v);
+ if ((priv->len - priv->off) < 1) {
+ return -E2BIG;
+ }
+ *v = priv->buffer[priv->off];
+ priv->off+=1;
+ //fprintf(stderr, "Deserializing bool end %d\n", priv->off);
+ return 0;
+}
+int ia_deserialize_buffer(struct iarchive *ia, const char *name,
+ struct buffer *b)
+{
+ struct buff_struct *priv = ia->priv;
+ int rc = ia_deserialize_int(ia, "len", &b->len);
+ if (rc < 0)
+ return rc;
+ if ((priv->len - priv->off) < b->len) {
+ return -E2BIG;
+ }
+ // set the buffer to null
+ if (b->len == -1) {
+ b->buff = NULL;
+ return rc;
+ }
+ b->buff = malloc(b->len);
+ if (!b->buff) {
+ return -ENOMEM;
+ }
+ memcpy(b->buff, priv->buffer+priv->off, b->len);
+ priv->off += b->len;
+ return 0;
+}
+int ia_deserialize_string(struct iarchive *ia, const char *name, char **s)
+{
+ struct buff_struct *priv = ia->priv;
+ int32_t len;
+ int rc = ia_deserialize_int(ia, "len", &len);
+ if (rc < 0)
+ return rc;
+ if ((priv->len - priv->off) < len) {
+ return -E2BIG;
+ }
+ if (len < 0) {
+ return -EINVAL;
+ }
+ *s = malloc(len+1);
+ if (!*s) {
+ return -ENOMEM;
+ }
+ memcpy(*s, priv->buffer+priv->off, len);
+ (*s)[len] = '\0';
+ priv->off += len;
+ return 0;
+}
+
+static struct iarchive ia_default = { STRUCT_INITIALIZER (start_record ,ia_start_record),
+ STRUCT_INITIALIZER (end_record ,ia_end_record), STRUCT_INITIALIZER (start_vector , ia_start_vector),
+ STRUCT_INITIALIZER (end_vector ,ia_end_vector), STRUCT_INITIALIZER (deserialize_Bool , ia_deserialize_bool),
+ STRUCT_INITIALIZER (deserialize_Int ,ia_deserialize_int),
+ STRUCT_INITIALIZER (deserialize_Long , ia_deserialize_long) ,
+ STRUCT_INITIALIZER (deserialize_Buffer, ia_deserialize_buffer),
+ STRUCT_INITIALIZER (deserialize_String, ia_deserialize_string) };
+
+static struct oarchive oa_default = { STRUCT_INITIALIZER (start_record , oa_start_record),
+ STRUCT_INITIALIZER (end_record , oa_end_record), STRUCT_INITIALIZER (start_vector , oa_start_vector),
+ STRUCT_INITIALIZER (end_vector , oa_end_vector), STRUCT_INITIALIZER (serialize_Bool , oa_serialize_bool),
+ STRUCT_INITIALIZER (serialize_Int , oa_serialize_int),
+ STRUCT_INITIALIZER (serialize_Long , oa_serialize_long) ,
+ STRUCT_INITIALIZER (serialize_Buffer , oa_serialize_buffer),
+ STRUCT_INITIALIZER (serialize_String , oa_serialize_string) };
+
+struct iarchive *create_buffer_iarchive(char *buffer, int len)
+{
+ struct iarchive *ia;
+ struct buff_struct *buff;
+ ia = malloc(sizeof(*ia));
+ if (!ia) return 0;
+ buff = malloc(sizeof(struct buff_struct));
+ if (!buff) {
+ free(ia);
+ return 0;
+ }
+ *ia = ia_default;
+ buff->off = 0;
+ buff->buffer = buffer;
+ buff->len = len;
+ ia->priv = buff;
+ return ia;
+}
+
+struct oarchive *create_buffer_oarchive()
+{
+ struct oarchive *oa;
+ struct buff_struct *buff;
+ oa = malloc(sizeof(*oa));
+ if (!oa) return 0;
+ buff = malloc(sizeof(struct buff_struct));
+ if (!buff) {
+ free(oa);
+ return 0;
+ }
+ *oa = oa_default;
+ buff->off = 0;
+ buff->buffer = malloc(128);
+ buff->len = 128;
+ oa->priv = buff;
+ return oa;
+}
+
+void close_buffer_iarchive(struct iarchive **ia)
+{
+ free((*ia)->priv);
+ free(*ia);
+ *ia = 0;
+}
+
+void close_buffer_oarchive(struct oarchive **oa, int free_buffer)
+{
+ if (free_buffer) {
+ struct buff_struct *buff = (struct buff_struct *)(*oa)->priv;
+ if (buff->buffer) {
+ free(buff->buffer);
+ }
+ }
+ free((*oa)->priv);
+ free(*oa);
+ *oa = 0;
+}
+
+char *get_buffer(struct oarchive *oa)
+{
+ struct buff_struct *buff = oa->priv;
+ return buff->buffer;
+}
+int get_buffer_len(struct oarchive *oa)
+{
+ struct buff_struct *buff = oa->priv;
+ return buff->off;
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/st_adaptor.c
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/st_adaptor.c b/zookeeper-client/zookeeper-client-c/src/st_adaptor.c
new file mode 100644
index 0000000..23573b3
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/st_adaptor.c
@@ -0,0 +1,113 @@
+/**
+ * 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.
+ */
+
+#ifndef DLL_EXPORT
+# define USE_STATIC_LIB
+#endif
+
+#include "zk_adaptor.h"
+#include <stdlib.h>
+#include <time.h>
+
+int zoo_lock_auth(zhandle_t *zh)
+{
+ return 0;
+}
+int zoo_unlock_auth(zhandle_t *zh)
+{
+ return 0;
+}
+int lock_buffer_list(buffer_head_t *l)
+{
+ return 0;
+}
+int unlock_buffer_list(buffer_head_t *l)
+{
+ return 0;
+}
+int lock_completion_list(completion_head_t *l)
+{
+ return 0;
+}
+int unlock_completion_list(completion_head_t *l)
+{
+ return 0;
+}
+struct sync_completion *alloc_sync_completion(void)
+{
+ return (struct sync_completion*)calloc(1, sizeof(struct sync_completion));
+}
+int wait_sync_completion(struct sync_completion *sc)
+{
+ return 0;
+}
+
+void free_sync_completion(struct sync_completion *sc)
+{
+ free(sc);
+}
+
+void notify_sync_completion(struct sync_completion *sc)
+{
+}
+
+int process_async(int outstanding_sync)
+{
+ return outstanding_sync == 0;
+}
+
+int adaptor_init(zhandle_t *zh)
+{
+ return 0;
+}
+
+void adaptor_finish(zhandle_t *zh){}
+
+void adaptor_destroy(zhandle_t *zh){}
+
+int flush_send_queue(zhandle_t *, int);
+
+int adaptor_send_queue(zhandle_t *zh, int timeout)
+{
+ return flush_send_queue(zh, timeout);
+}
+
+int32_t inc_ref_counter(zhandle_t* zh,int i)
+{
+ zh->ref_counter+=(i<0?-1:(i>0?1:0));
+ return zh->ref_counter;
+}
+
+int32_t get_xid()
+{
+ static int32_t xid = -1;
+ if (xid == -1) {
+ xid = time(0);
+ }
+ return xid++;
+}
+
+int enter_critical(zhandle_t* zh)
+{
+ return 0;
+}
+
+int leave_critical(zhandle_t* zh)
+{
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/winport.c
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/winport.c b/zookeeper-client/zookeeper-client-c/src/winport.c
new file mode 100644
index 0000000..3592ea1
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/winport.c
@@ -0,0 +1,302 @@
+/**
+ * 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.
+ */
+
+#ifdef WIN32
+#include "winport.h"
+#include <stdlib.h>
+#include <stdint.h> /* for int64_t */
+#include <winsock2.h> /* must always be included before ws2tcpip.h */
+#include <ws2tcpip.h> /* for SOCKET */
+
+int pthread_mutex_lock(pthread_mutex_t* _mutex ){
+ int rc = WaitForSingleObject( *_mutex, // handle to mutex
+ INFINITE); // no time-out interval
+ return ((rc == WAIT_OBJECT_0) ? 0: rc);
+}
+
+int pthread_mutex_unlock( pthread_mutex_t* _mutex ){
+ int rc = ReleaseMutex(*_mutex);
+ return ((rc != 0)? 0: GetLastError());
+}
+
+int pthread_mutex_init(pthread_mutex_t* _mutex, void* ignoredAttr){
+ //use CreateMutex as we are using the HANDLES in pthread_cond
+ *_mutex = CreateMutex(
+ NULL, // default security attributes
+ FALSE, // initially not owned
+ NULL); // unnamed mutex
+ return ((*_mutex == NULL) ? GetLastError() : 0);
+}
+
+int pthread_mutex_destroy(pthread_mutex_t* _mutex)
+{
+ int rc = CloseHandle(*_mutex);
+ return ((rc != 0)? 0: GetLastError());
+}
+
+int pthread_create(pthread_t *thread, const pthread_attr_t *attr, unsigned (__stdcall* start_routine)(void* a), void *arg)
+{
+ int _intThreadId;
+ (*thread).thread_handle = (HANDLE)_beginthreadex( NULL, 0, start_routine , arg, 0, (unsigned int*)&_intThreadId );
+ (*thread).thread_id = _intThreadId;
+ return (((*thread).thread_handle == 0 ) ? errno : 0 );
+}
+
+
+int pthread_equal(pthread_t t1, pthread_t t2){
+//Is there a better way to do this? GetThreadId(handle) is only supported Windows 2003 n above.
+ return ((t1.thread_id == t2.thread_id) ? 1:0);
+}
+
+pthread_t pthread_self(){
+ pthread_t thread_self;
+ thread_self.thread_handle = GetCurrentThread();
+ thread_self.thread_id = GetCurrentThreadId();
+ return thread_self;
+}
+
+int pthread_join(pthread_t _thread, void** ignore)
+{
+ int rc = WaitForSingleObject( _thread.thread_handle, INFINITE );
+ return ((rc == WAIT_OBJECT_0) ? 0: rc);
+}
+
+int pthread_detach(pthread_t _thread)
+{
+ int rc = CloseHandle(_thread.thread_handle) ;
+ return (rc != 0) ? 0: GetLastError();
+}
+
+void pthread_mutexattr_init(pthread_mutexattr_t* ignore){}
+void pthread_mutexattr_settype(pthread_mutexattr_t* ingore_attr, int ignore){}
+void pthread_mutexattr_destroy(pthread_mutexattr_t* ignore_attr){}
+
+int
+pthread_cond_init (pthread_cond_t *cv,
+ const pthread_condattr_t * ignore)
+{
+ cv->waiters_count_ = 0;
+ cv->was_broadcast_ = 0;
+ cv->sema_ = CreateSemaphore (NULL, // no security
+ 0, // initially 0
+ 0x7fffffff, // max count
+ NULL); // unnamed
+ if (cv->sema_ == NULL )
+ return GetLastError();
+ InitializeCriticalSection (&cv->waiters_count_lock_);
+ cv->waiters_done_ = CreateEvent (NULL, // no security
+ FALSE, // auto-reset
+ FALSE, // non-signaled initially
+ NULL); // unnamed
+ return (cv->waiters_done_ == NULL) ? GetLastError() : 0;
+
+}
+
+
+int pthread_cond_destroy(pthread_cond_t *cond)
+{
+ CloseHandle( cond->sema_);
+ DeleteCriticalSection(&cond->waiters_count_lock_);
+ return (CloseHandle( cond->waiters_done_ ) == 0)? GetLastError(): 0 ;
+}
+
+
+int
+pthread_cond_signal (pthread_cond_t *cv)
+{
+ int have_waiters;
+ EnterCriticalSection (& (cv->waiters_count_lock_));
+ have_waiters = cv->waiters_count_ > 0;
+ LeaveCriticalSection (&cv->waiters_count_lock_);
+
+ // If there aren't any waiters, then this is a no-op.
+ if (have_waiters){
+ return (ReleaseSemaphore (cv->sema_, 1, 0) == 0 ) ? GetLastError() : 0 ;
+ }else
+ return 0;
+}
+
+
+int
+pthread_cond_broadcast (pthread_cond_t *cv)
+{
+ // This is needed to ensure that <waiters_count_> and <was_broadcast_> are
+ // consistent relative to each other.
+ int have_waiters = 0;
+ EnterCriticalSection (&cv->waiters_count_lock_);
+
+ if (cv->waiters_count_ > 0) {
+ // We are broadcasting, even if there is just one waiter...
+ // Record that we are broadcasting, which helps optimize
+ // <pthread_cond_wait> for the non-broadcast case.
+ cv->was_broadcast_ = 1;
+ have_waiters = 1;
+ }
+
+ if (have_waiters) {
+ // Wake up all the waiters atomically.
+ ReleaseSemaphore (cv->sema_, cv->waiters_count_, 0);
+
+ LeaveCriticalSection (&cv->waiters_count_lock_);
+
+ // Wait for all the awakened threads to acquire the counting
+ // semaphore.
+ WaitForSingleObject (cv->waiters_done_, INFINITE);
+ // This assignment is okay, even without the <waiters_count_lock_> held
+ // because no other waiter threads can wake up to access it.
+ cv->was_broadcast_ = 0;
+ }
+ else
+ LeaveCriticalSection (&cv->waiters_count_lock_);
+}
+
+
+int
+pthread_cond_wait (pthread_cond_t *cv,
+ pthread_mutex_t *external_mutex)
+{
+ int last_waiter;
+ // Avoid race conditions.
+ EnterCriticalSection (&cv->waiters_count_lock_);
+ cv->waiters_count_++;
+ LeaveCriticalSection (&cv->waiters_count_lock_);
+
+ // This call atomically releases the mutex and waits on the
+ // semaphore until <pthread_cond_signal> or <pthread_cond_broadcast>
+ // are called by another thread.
+ SignalObjectAndWait (*external_mutex, cv->sema_, INFINITE, FALSE);
+
+ // Reacquire lock to avoid race conditions.
+ EnterCriticalSection (&cv->waiters_count_lock_);
+
+ // We're no longer waiting...
+ cv->waiters_count_--;
+
+ // Check to see if we're the last waiter after <pthread_cond_broadcast>.
+ last_waiter = cv->was_broadcast_ && cv->waiters_count_ == 0;
+
+ LeaveCriticalSection (&cv->waiters_count_lock_);
+
+ // If we're the last waiter thread during this particular broadcast
+ // then let all the other threads proceed.
+ if (last_waiter)
+ // This call atomically signals the <waiters_done_> event and waits until
+ // it can acquire the <external_mutex>. This is required to ensure fairness.
+ SignalObjectAndWait (cv->waiters_done_, *external_mutex, INFINITE, FALSE);
+ else
+ // Always regain the external mutex since that's the guarantee we
+ // give to our callers.
+ WaitForSingleObject (*external_mutex, INFINITE);
+}
+
+int pthread_key_create(pthread_key_t *key, void (*destructor)(void *) )
+{
+ int result = 0;
+ pthread_key_t* newkey;
+
+ if ((newkey = (pthread_key_t*) calloc (1, sizeof (pthread_key_t))) == NULL)
+ {
+ result = ENOMEM;
+ }
+ else if ((newkey->key = TlsAlloc ()) == TLS_OUT_OF_INDEXES)
+ {
+ result = EAGAIN;
+ free (newkey);
+ newkey = NULL;
+ }
+ else if (destructor != NULL)
+ {
+ //--we have to store the function pointer for destructor, so that we can call it
+ //--to free up the user allocated storage--
+ newkey->destructor = destructor;
+ }
+ key = newkey;
+ return (result);
+}
+
+int pthread_key_delete(pthread_key_t key)
+{
+ int rc = 0;
+ LPVOID lpvData = TlsGetValue(key.key);
+ rc = TlsFree (key.key);
+ rc = (rc != 0 ) ? 0 : GetLastError();
+ if (key.destructor != NULL && lpvData != 0){
+ key.destructor(lpvData); //we take control of calling destructor, instead of calling it on thread exit.
+ }
+ free (&key);
+ return (rc);
+}
+
+void *pthread_getspecific(pthread_key_t key)
+{
+ LPVOID lpvData = TlsGetValue(key.key);
+ if ((lpvData == 0) && (GetLastError() != ERROR_SUCCESS))
+ return NULL;
+ else
+ return lpvData;
+}
+
+int pthread_setspecific(pthread_key_t key, const void *value)
+{
+ int rc = TlsSetValue (key.key, value);
+ return ((rc != 0 ) ? 0 : GetLastError());
+}
+
+int gettimeofday(struct timeval *tp, void *tzp) {
+ int64_t now = 0;
+ if (tzp != 0) { errno = EINVAL; return -1; }
+ GetSystemTimeAsFileTime( (LPFILETIME)&now );
+ tp->tv_sec = (long)(now / 10000000 - 11644473600LL);
+ tp->tv_usec = (now / 10) % 1000000;
+ return 0;
+}
+
+int close(SOCKET fd) {
+ return closesocket(fd);
+}
+
+int Win32WSAStartup()
+{
+ WORD wVersionRq;
+ WSADATA wsaData;
+ int err;
+
+ wVersionRq = MAKEWORD(2,0);
+ err = WSAStartup(wVersionRq, &wsaData);
+ if (err != 0)
+ return 1;
+
+ // confirm the version information
+ if ((LOBYTE(wsaData.wVersion) != 2) ||
+ (HIBYTE(wsaData.wVersion) != 0))
+ {
+ Win32WSACleanup();
+ return 1;
+ }
+ return 0;
+}
+
+void Win32WSACleanup()
+{
+ WSACleanup();
+}
+
+#endif //WIN32
+
+
+
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/winport.h
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/winport.h b/zookeeper-client/zookeeper-client-c/src/winport.h
new file mode 100644
index 0000000..da6028c
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/winport.h
@@ -0,0 +1,138 @@
+/**
+ * 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.
+ */
+
+/**
+ * This header file is to port pthread lib , sockets and other utility methods on windows.
+ * Specifically the threads function, mutexes, keys, and socket initialization.
+ */
+
+#ifndef WINPORT_H_
+#define WINPORT_H_
+
+#ifdef WIN32
+#include "winconfig.h"
+
+#define _WINSOCK_DEPRECATED_NO_WARNINGS
+#include <winsock2.h> /* must always be included before ws2tcpip.h */
+#include <ws2tcpip.h> /* for struct sock_addr used in zookeeper.h */
+
+/* POSIX names are deprecated, use ISO conformant names instead. */
+#define strdup _strdup
+#define getcwd _getcwd
+#define getpid _getpid
+
+/* Windows "secure" versions of POSIX reentrant functions */
+#define strtok_r strtok_s
+#define localtime_r(a,b) localtime_s(b,a)
+
+/* After this version of MSVC, snprintf became a defined function,
+ and so cannot be redefined, nor can #ifndef be used to guard it. */
+#if ((defined(_MSC_VER) && _MSC_VER < 1900) || !defined(_MSC_VER))
+#define snprintf _snprintf
+#endif
+
+
+#include <errno.h>
+#include <process.h>
+#include <stdint.h> /* for int64_t */
+#include <stdlib.h>
+#include <malloc.h>
+
+
+typedef int ssize_t;
+typedef HANDLE pthread_mutex_t;
+
+struct pthread_t_
+{
+ HANDLE thread_handle;
+ DWORD thread_id;
+};
+
+typedef struct pthread_t_ pthread_t;
+typedef int pthread_mutexattr_t;
+typedef int pthread_condattr_t;
+typedef int pthread_attr_t;
+#define PTHREAD_MUTEX_RECURSIVE 0
+
+int pthread_mutex_lock(pthread_mutex_t* _mutex );
+int pthread_mutex_unlock( pthread_mutex_t* _mutex );
+int pthread_mutex_init(pthread_mutex_t* _mutex, void* ignoredAttr);
+int pthread_mutex_destroy(pthread_mutex_t* _mutex);
+int pthread_create(pthread_t *thread, const pthread_attr_t *attr, unsigned (__stdcall* start_routine)(void* a), void *arg);
+int pthread_equal(pthread_t t1, pthread_t t2);
+pthread_t pthread_self();
+int pthread_join(pthread_t _thread, void** ignore);
+int pthread_detach(pthread_t _thread);
+
+void pthread_mutexattr_init(pthread_mutexattr_t* ignore);
+void pthread_mutexattr_settype(pthread_mutexattr_t* ingore_attr, int ignore);
+void pthread_mutexattr_destroy(pthread_mutexattr_t* ignore_attr);
+
+
+// http://www.cs.wustl.edu/~schmidt/win32-cv-1.html
+
+typedef struct
+{
+ int waiters_count_;
+ // Number of waiting threads.
+
+ CRITICAL_SECTION waiters_count_lock_;
+ // Serialize access to <waiters_count_>.
+
+ HANDLE sema_;
+ // Semaphore used to queue up threads waiting for the condition to
+ // become signaled.
+
+ HANDLE waiters_done_;
+ // An auto-reset event used by the broadcast/signal thread to wait
+ // for all the waiting thread(s) to wake up and be released from the
+ // semaphore.
+
+ size_t was_broadcast_;
+ // Keeps track of whether we were broadcasting or signaling. This
+ // allows us to optimize the code if we're just signaling.
+}pthread_cond_t;
+
+int pthread_cond_init (pthread_cond_t *cv,const pthread_condattr_t * ignore);
+int pthread_cond_destroy(pthread_cond_t *cond);
+int pthread_cond_signal (pthread_cond_t *cv);
+int pthread_cond_broadcast (pthread_cond_t *cv);
+int pthread_cond_wait (pthread_cond_t *cv, pthread_mutex_t *external_mutex);
+
+
+struct pthread_key_t_
+{
+ DWORD key;
+ void (*destructor) (void *);
+};
+
+typedef struct pthread_key_t_ pthread_key_t;
+int pthread_key_create(pthread_key_t *key, void (*destructor)(void *) );
+int pthread_key_delete(pthread_key_t key);
+void *pthread_getspecific(pthread_key_t key);
+int pthread_setspecific(pthread_key_t key, const void *value);
+
+int gettimeofday(struct timeval *tp, void *tzp);
+int close(SOCKET fd);
+int Win32WSAStartup();
+void Win32WSACleanup();
+#endif //WIN32
+
+
+
+#endif //WINPORT_H_
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/zk_adaptor.h
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/zk_adaptor.h b/zookeeper-client/zookeeper-client-c/src/zk_adaptor.h
new file mode 100644
index 0000000..b990779
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/zk_adaptor.h
@@ -0,0 +1,276 @@
+/**
+ * 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.
+ */
+
+#ifndef ZK_ADAPTOR_H_
+#define ZK_ADAPTOR_H_
+#include <zookeeper.jute.h>
+#ifdef THREADED
+#ifndef WIN32
+#include <pthread.h>
+#else
+#include "winport.h"
+#endif
+#endif
+#include "zookeeper.h"
+#include "zk_hashtable.h"
+
+/* predefined xid's values recognized as special by the server */
+#define WATCHER_EVENT_XID -1
+#define PING_XID -2
+#define AUTH_XID -4
+#define SET_WATCHES_XID -8
+
+/* zookeeper state constants */
+#define EXPIRED_SESSION_STATE_DEF -112
+#define AUTH_FAILED_STATE_DEF -113
+#define CONNECTING_STATE_DEF 1
+#define ASSOCIATING_STATE_DEF 2
+#define CONNECTED_STATE_DEF 3
+#define NOTCONNECTED_STATE_DEF 999
+
+/* zookeeper event type constants */
+#define CREATED_EVENT_DEF 1
+#define DELETED_EVENT_DEF 2
+#define CHANGED_EVENT_DEF 3
+#define CHILD_EVENT_DEF 4
+#define SESSION_EVENT_DEF -1
+#define NOTWATCHING_EVENT_DEF -2
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct _buffer_list;
+struct _completion_list;
+
+typedef struct _buffer_head {
+ struct _buffer_list *volatile head;
+ struct _buffer_list *last;
+#ifdef THREADED
+ pthread_mutex_t lock;
+#endif
+} buffer_head_t;
+
+typedef struct _completion_head {
+ struct _completion_list *volatile head;
+ struct _completion_list *last;
+#ifdef THREADED
+ pthread_cond_t cond;
+ pthread_mutex_t lock;
+#endif
+} completion_head_t;
+
+int lock_buffer_list(buffer_head_t *l);
+int unlock_buffer_list(buffer_head_t *l);
+int lock_completion_list(completion_head_t *l);
+int unlock_completion_list(completion_head_t *l);
+
+struct sync_completion {
+ int rc;
+ union {
+ struct {
+ char *str;
+ int str_len;
+ } str;
+ struct Stat stat;
+ struct {
+ char *buffer;
+ int buff_len;
+ struct Stat stat;
+ } data;
+ struct {
+ struct ACL_vector acl;
+ struct Stat stat;
+ } acl;
+ struct String_vector strs2;
+ struct {
+ struct String_vector strs2;
+ struct Stat stat2;
+ } strs_stat;
+ } u;
+ int complete;
+#ifdef THREADED
+ pthread_cond_t cond;
+ pthread_mutex_t lock;
+#endif
+};
+
+typedef struct _auth_info {
+ int state; /* 0=>inactive, >0 => active */
+ char* scheme;
+ struct buffer auth;
+ void_completion_t completion;
+ const char* data;
+ struct _auth_info *next;
+} auth_info;
+
+/**
+ * This structure represents a packet being read or written.
+ */
+typedef struct _buffer_list {
+ char *buffer;
+ int len; /* This represents the length of sizeof(header) + length of buffer */
+ int curr_offset; /* This is the offset into the header followed by offset into the buffer */
+ struct _buffer_list *next;
+} buffer_list_t;
+
+/* the size of connect request */
+#define HANDSHAKE_REQ_SIZE 44
+/* connect request */
+struct connect_req {
+ int32_t protocolVersion;
+ int64_t lastZxidSeen;
+ int32_t timeOut;
+ int64_t sessionId;
+ int32_t passwd_len;
+ char passwd[16];
+};
+
+/* the connect response */
+struct prime_struct {
+ int32_t len;
+ int32_t protocolVersion;
+ int32_t timeOut;
+ int64_t sessionId;
+ int32_t passwd_len;
+ char passwd[16];
+};
+
+#ifdef THREADED
+/* this is used by mt_adaptor internally for thread management */
+struct adaptor_threads {
+ pthread_t io;
+ pthread_t completion;
+ int threadsToWait; // barrier
+ pthread_cond_t cond; // barrier's conditional
+ pthread_mutex_t lock; // ... and a lock
+ pthread_mutex_t zh_lock; // critical section lock
+#ifdef WIN32
+ SOCKET self_pipe[2];
+#else
+ int self_pipe[2];
+#endif
+};
+#endif
+
+/** the auth list for adding auth */
+typedef struct _auth_list_head {
+ auth_info *auth;
+#ifdef THREADED
+ pthread_mutex_t lock;
+#endif
+} auth_list_head_t;
+
+/**
+ * This structure represents the connection to zookeeper.
+ */
+
+struct _zhandle {
+#ifdef WIN32
+ SOCKET fd; /* the descriptor used to talk to zookeeper */
+#else
+ int fd; /* the descriptor used to talk to zookeeper */
+#endif
+ char *hostname; /* the hostname of zookeeper */
+ struct sockaddr_storage *addrs; /* the addresses that correspond to the hostname */
+ int addrs_count; /* The number of addresses in the addrs array */
+ watcher_fn watcher; /* the registered watcher */
+ struct timeval last_recv; /* The time that the last message was received */
+ struct timeval last_send; /* The time that the last message was sent */
+ struct timeval last_ping; /* The time that the last PING was sent */
+ struct timeval next_deadline; /* The time of the next deadline */
+ int recv_timeout; /* The maximum amount of time that can go by without
+ receiving anything from the zookeeper server */
+ buffer_list_t *input_buffer; /* the current buffer being read in */
+ buffer_head_t to_process; /* The buffers that have been read and are ready to be processed. */
+ buffer_head_t to_send; /* The packets queued to send */
+ completion_head_t sent_requests; /* The outstanding requests */
+ completion_head_t completions_to_process; /* completions that are ready to run */
+ int connect_index; /* The index of the address to connect to */
+ clientid_t client_id;
+ long long last_zxid;
+ int outstanding_sync; /* Number of outstanding synchronous requests */
+ struct _buffer_list primer_buffer; /* The buffer used for the handshake at the start of a connection */
+ struct prime_struct primer_storage; /* the connect response */
+ char primer_storage_buffer[40]; /* the true size of primer_storage */
+ volatile int state;
+ void *context;
+ auth_list_head_t auth_h; /* authentication data list */
+ /* zookeeper_close is not reentrant because it de-allocates the zhandler.
+ * This guard variable is used to defer the destruction of zhandle till
+ * right before top-level API call returns to the caller */
+ int32_t ref_counter;
+ volatile int close_requested;
+ void *adaptor_priv;
+ /* Used for debugging only: non-zero value indicates the time when the zookeeper_process
+ * call returned while there was at least one unprocessed server response
+ * available in the socket recv buffer */
+ struct timeval socket_readable;
+
+ zk_hashtable* active_node_watchers;
+ zk_hashtable* active_exist_watchers;
+ zk_hashtable* active_child_watchers;
+ /** used for chroot path at the client side **/
+ char *chroot;
+};
+
+
+int adaptor_init(zhandle_t *zh);
+void adaptor_finish(zhandle_t *zh);
+void adaptor_destroy(zhandle_t *zh);
+struct sync_completion *alloc_sync_completion(void);
+int wait_sync_completion(struct sync_completion *sc);
+void free_sync_completion(struct sync_completion *sc);
+void notify_sync_completion(struct sync_completion *sc);
+int adaptor_send_queue(zhandle_t *zh, int timeout);
+int process_async(int outstanding_sync);
+void process_completions(zhandle_t *zh);
+int flush_send_queue(zhandle_t*zh, int timeout);
+char* sub_string(zhandle_t *zh, const char* server_path);
+void free_duplicate_path(const char* free_path, const char* path);
+int zoo_lock_auth(zhandle_t *zh);
+int zoo_unlock_auth(zhandle_t *zh);
+
+// critical section guards
+int enter_critical(zhandle_t* zh);
+int leave_critical(zhandle_t* zh);
+// zhandle object reference counting
+void api_prolog(zhandle_t* zh);
+int api_epilog(zhandle_t *zh, int rc);
+int32_t get_xid();
+// returns the new value of the ref counter
+int32_t inc_ref_counter(zhandle_t* zh,int i);
+
+#ifdef THREADED
+// atomic post-increment
+int32_t fetch_and_add(volatile int32_t* operand, int incr);
+// in mt mode process session event asynchronously by the completion thread
+#define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
+#else
+// in single-threaded mode process session event immediately
+//#define PROCESS_SESSION_EVENT(zh,newstate) deliverWatchers(zh,ZOO_SESSION_EVENT,newstate,0)
+#define PROCESS_SESSION_EVENT(zh,newstate) queue_session_event(zh,newstate)
+#endif
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*ZK_ADAPTOR_H_*/
+
+
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/zk_hashtable.c
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/zk_hashtable.c b/zookeeper-client/zookeeper-client-c/src/zk_hashtable.c
new file mode 100644
index 0000000..0efc5aa
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/zk_hashtable.c
@@ -0,0 +1,337 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "zk_hashtable.h"
+#include "zk_adaptor.h"
+#include "hashtable/hashtable.h"
+#include "hashtable/hashtable_itr.h"
+#include <string.h>
+#include <stdlib.h>
+#include <assert.h>
+
+typedef struct _watcher_object {
+ watcher_fn watcher;
+ void* context;
+ struct _watcher_object* next;
+} watcher_object_t;
+
+
+struct _zk_hashtable {
+ struct hashtable* ht;
+};
+
+struct watcher_object_list {
+ watcher_object_t* head;
+};
+
+/* the following functions are for testing only */
+typedef struct hashtable hashtable_impl;
+
+hashtable_impl* getImpl(zk_hashtable* ht){
+ return ht->ht;
+}
+
+watcher_object_t* getFirstWatcher(zk_hashtable* ht,const char* path)
+{
+ watcher_object_list_t* wl=hashtable_search(ht->ht,(void*)path);
+ if(wl!=0)
+ return wl->head;
+ return 0;
+}
+/* end of testing functions */
+
+watcher_object_t* clone_watcher_object(watcher_object_t* wo)
+{
+ watcher_object_t* res=calloc(1,sizeof(watcher_object_t));
+ assert(res);
+ res->watcher=wo->watcher;
+ res->context=wo->context;
+ return res;
+}
+
+static unsigned int string_hash_djb2(void *str)
+{
+ unsigned int hash = 5381;
+ int c;
+ const char* cstr = (const char*)str;
+ while ((c = *cstr++))
+ hash = ((hash << 5) + hash) + c; /* hash * 33 + c */
+
+ return hash;
+}
+
+static int string_equal(void *key1,void *key2)
+{
+ return strcmp((const char*)key1,(const char*)key2)==0;
+}
+
+static watcher_object_t* create_watcher_object(watcher_fn watcher,void* ctx)
+{
+ watcher_object_t* wo=calloc(1,sizeof(watcher_object_t));
+ assert(wo);
+ wo->watcher=watcher;
+ wo->context=ctx;
+ return wo;
+}
+
+static watcher_object_list_t* create_watcher_object_list(watcher_object_t* head)
+{
+ watcher_object_list_t* wl=calloc(1,sizeof(watcher_object_list_t));
+ assert(wl);
+ wl->head=head;
+ return wl;
+}
+
+static void destroy_watcher_object_list(watcher_object_list_t* list)
+{
+ watcher_object_t* e = NULL;
+
+ if(list==0)
+ return;
+ e=list->head;
+ while(e!=0){
+ watcher_object_t* this=e;
+ e=e->next;
+ free(this);
+ }
+ free(list);
+}
+
+zk_hashtable* create_zk_hashtable()
+{
+ struct _zk_hashtable *ht=calloc(1,sizeof(struct _zk_hashtable));
+ assert(ht);
+ ht->ht=create_hashtable(32,string_hash_djb2,string_equal);
+ return ht;
+}
+
+static void do_clean_hashtable(zk_hashtable* ht)
+{
+ struct hashtable_itr *it;
+ int hasMore;
+ if(hashtable_count(ht->ht)==0)
+ return;
+ it=hashtable_iterator(ht->ht);
+ do {
+ watcher_object_list_t* w=hashtable_iterator_value(it);
+ destroy_watcher_object_list(w);
+ hasMore=hashtable_iterator_remove(it);
+ } while(hasMore);
+ free(it);
+}
+
+void destroy_zk_hashtable(zk_hashtable* ht)
+{
+ if(ht!=0){
+ do_clean_hashtable(ht);
+ hashtable_destroy(ht->ht,0);
+ free(ht);
+ }
+}
+
+// searches for a watcher object instance in a watcher object list;
+// two watcher objects are equal if their watcher function and context pointers
+// are equal
+static watcher_object_t* search_watcher(watcher_object_list_t** wl,watcher_object_t* wo)
+{
+ watcher_object_t* wobj=(*wl)->head;
+ while(wobj!=0){
+ if(wobj->watcher==wo->watcher && wobj->context==wo->context)
+ return wobj;
+ wobj=wobj->next;
+ }
+ return 0;
+}
+
+static int add_to_list(watcher_object_list_t **wl, watcher_object_t *wo,
+ int clone)
+{
+ if (search_watcher(wl, wo)==0) {
+ watcher_object_t* cloned=wo;
+ if (clone) {
+ cloned = clone_watcher_object(wo);
+ assert(cloned);
+ }
+ cloned->next = (*wl)->head;
+ (*wl)->head = cloned;
+ return 1;
+ } else if (!clone) {
+ // If it's here and we aren't supposed to clone, we must destroy
+ free(wo);
+ }
+ return 0;
+}
+
+static int do_insert_watcher_object(zk_hashtable *ht, const char *path, watcher_object_t* wo)
+{
+ int res=1;
+ watcher_object_list_t* wl;
+
+ wl=hashtable_search(ht->ht,(void*)path);
+ if(wl==0){
+ int res;
+ /* inserting a new path element */
+ res=hashtable_insert(ht->ht,strdup(path),create_watcher_object_list(wo));
+ assert(res);
+ }else{
+ /*
+ * Path already exists; check if the watcher already exists.
+ * Don't clone the watcher since it's allocated on the heap --- avoids
+ * a memory leak and saves a clone operation (calloc + copy).
+ */
+ res = add_to_list(&wl, wo, 0);
+ }
+ return res;
+}
+
+
+char **collect_keys(zk_hashtable *ht, int *count)
+{
+ char **list;
+ struct hashtable_itr *it;
+ int i;
+
+ *count = hashtable_count(ht->ht);
+ list = calloc(*count, sizeof(char*));
+ it=hashtable_iterator(ht->ht);
+ for(i = 0; i < *count; i++) {
+ list[i] = strdup(hashtable_iterator_key(it));
+ hashtable_iterator_advance(it);
+ }
+ free(it);
+ return list;
+}
+
+static int insert_watcher_object(zk_hashtable *ht, const char *path,
+ watcher_object_t* wo)
+{
+ int res;
+ res=do_insert_watcher_object(ht,path,wo);
+ return res;
+}
+
+static void copy_watchers(watcher_object_list_t *from, watcher_object_list_t *to, int clone)
+{
+ watcher_object_t* wo=from->head;
+ while(wo){
+ watcher_object_t *next = wo->next;
+ add_to_list(&to, wo, clone);
+ wo=next;
+ }
+}
+
+static void copy_table(zk_hashtable *from, watcher_object_list_t *to) {
+ struct hashtable_itr *it;
+ int hasMore;
+ if(hashtable_count(from->ht)==0)
+ return;
+ it=hashtable_iterator(from->ht);
+ do {
+ watcher_object_list_t *w = hashtable_iterator_value(it);
+ copy_watchers(w, to, 1);
+ hasMore=hashtable_iterator_advance(it);
+ } while(hasMore);
+ free(it);
+}
+
+static void collect_session_watchers(zhandle_t *zh,
+ watcher_object_list_t **list)
+{
+ copy_table(zh->active_node_watchers, *list);
+ copy_table(zh->active_exist_watchers, *list);
+ copy_table(zh->active_child_watchers, *list);
+}
+
+static void add_for_event(zk_hashtable *ht, char *path, watcher_object_list_t **list)
+{
+ watcher_object_list_t* wl;
+ wl = (watcher_object_list_t*)hashtable_remove(ht->ht, path);
+ if (wl) {
+ copy_watchers(wl, *list, 0);
+ // Since we move, not clone the watch_objects, we just need to free the
+ // head pointer
+ free(wl);
+ }
+}
+
+static void do_foreach_watcher(watcher_object_t* wo,zhandle_t* zh,
+ const char* path,int type,int state)
+{
+ // session event's don't have paths
+ const char *client_path =
+ (type != ZOO_SESSION_EVENT ? sub_string(zh, path) : path);
+ while(wo!=0){
+ wo->watcher(zh,type,state,client_path,wo->context);
+ wo=wo->next;
+ }
+ free_duplicate_path(client_path, path);
+}
+
+watcher_object_list_t *collectWatchers(zhandle_t *zh,int type, char *path)
+{
+ struct watcher_object_list *list = create_watcher_object_list(0);
+
+ if(type==ZOO_SESSION_EVENT){
+ watcher_object_t defWatcher;
+ defWatcher.watcher=zh->watcher;
+ defWatcher.context=zh->context;
+ add_to_list(&list, &defWatcher, 1);
+ collect_session_watchers(zh, &list);
+ return list;
+ }
+ switch(type){
+ case CREATED_EVENT_DEF:
+ case CHANGED_EVENT_DEF:
+ // look up the watchers for the path and move them to a delivery list
+ add_for_event(zh->active_node_watchers,path,&list);
+ add_for_event(zh->active_exist_watchers,path,&list);
+ break;
+ case CHILD_EVENT_DEF:
+ // look up the watchers for the path and move them to a delivery list
+ add_for_event(zh->active_child_watchers,path,&list);
+ break;
+ case DELETED_EVENT_DEF:
+ // look up the watchers for the path and move them to a delivery list
+ add_for_event(zh->active_node_watchers,path,&list);
+ add_for_event(zh->active_exist_watchers,path,&list);
+ add_for_event(zh->active_child_watchers,path,&list);
+ break;
+ }
+ return list;
+}
+
+void deliverWatchers(zhandle_t *zh, int type,int state, char *path, watcher_object_list_t **list)
+{
+ if (!list || !(*list)) return;
+ do_foreach_watcher((*list)->head, zh, path, type, state);
+ destroy_watcher_object_list(*list);
+ *list = 0;
+}
+
+void activateWatcher(zhandle_t *zh, watcher_registration_t* reg, int rc)
+{
+ if(reg){
+ /* in multithreaded lib, this code is executed
+ * by the IO thread */
+ zk_hashtable *ht = reg->checker(zh, rc);
+ if(ht){
+ insert_watcher_object(ht,reg->path,
+ create_watcher_object(reg->watcher, reg->context));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/zk_hashtable.h
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/zk_hashtable.h b/zookeeper-client/zookeeper-client-c/src/zk_hashtable.h
new file mode 100644
index 0000000..31109c1
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/zk_hashtable.h
@@ -0,0 +1,69 @@
+/**
+ * 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.
+ */
+
+#ifndef ZK_HASHTABLE_H_
+#define ZK_HASHTABLE_H_
+
+#include <zookeeper.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+ typedef struct watcher_object_list watcher_object_list_t;
+typedef struct _zk_hashtable zk_hashtable;
+
+/**
+ * The function must return a non-zero value if the watcher object can be activated
+ * as a result of the server response. Normally, a watch can only be activated
+ * if the server returns a success code (ZOK). However in the case when zoo_exists()
+ * returns a ZNONODE code the watcher should be activated nevertheless.
+ */
+typedef zk_hashtable *(*result_checker_fn)(zhandle_t *, int rc);
+
+/**
+ * A watcher object gets temporarily stored with the completion entry until
+ * the server response comes back at which moment the watcher object is moved
+ * to the active watchers map.
+ */
+typedef struct _watcher_registration {
+ watcher_fn watcher;
+ void* context;
+ result_checker_fn checker;
+ const char* path;
+} watcher_registration_t;
+
+zk_hashtable* create_zk_hashtable();
+void destroy_zk_hashtable(zk_hashtable* ht);
+
+char **collect_keys(zk_hashtable *ht, int *count);
+
+/**
+ * check if the completion has a watcher object associated
+ * with it. If it does, move the watcher object to the map of
+ * active watchers (only if the checker allows to do so)
+ */
+ void activateWatcher(zhandle_t *zh, watcher_registration_t* reg, int rc);
+ watcher_object_list_t *collectWatchers(zhandle_t *zh,int type, char *path);
+ void deliverWatchers(zhandle_t *zh, int type, int state, char *path, struct watcher_object_list **list);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*ZK_HASHTABLE_H_*/
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/5694559d/zookeeper-client/zookeeper-client-c/src/zk_log.c
----------------------------------------------------------------------
diff --git a/zookeeper-client/zookeeper-client-c/src/zk_log.c b/zookeeper-client/zookeeper-client-c/src/zk_log.c
new file mode 100644
index 0000000..6b4fdfa
--- /dev/null
+++ b/zookeeper-client/zookeeper-client-c/src/zk_log.c
@@ -0,0 +1,180 @@
+/**
+ * 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.
+ */
+
+#if !defined(DLL_EXPORT) && !defined(USE_STATIC_LIB)
+# define USE_STATIC_LIB
+#endif
+
+#include "zookeeper_log.h"
+#ifndef WIN32
+#include <unistd.h>
+#else
+typedef DWORD pid_t;
+#include <process.h> /* for getpid */
+#endif
+
+#include <stdarg.h>
+#include <time.h>
+
+#define TIME_NOW_BUF_SIZE 1024
+#define FORMAT_LOG_BUF_SIZE 4096
+
+#ifdef THREADED
+#ifndef WIN32
+#include <pthread.h>
+#else
+#include "winport.h"
+#endif
+
+static pthread_key_t time_now_buffer;
+static pthread_key_t format_log_msg_buffer;
+
+void freeBuffer(void* p){
+ if(p) free(p);
+}
+
+__attribute__((constructor)) void prepareTSDKeys() {
+ pthread_key_create (&time_now_buffer, freeBuffer);
+ pthread_key_create (&format_log_msg_buffer, freeBuffer);
+}
+
+char* getTSData(pthread_key_t key,int size){
+ char* p=pthread_getspecific(key);
+ if(p==0){
+ int res;
+ p=calloc(1,size);
+ res=pthread_setspecific(key,p);
+ if(res!=0){
+ fprintf(stderr,"Failed to set TSD key: %d",res);
+ }
+ }
+ return p;
+}
+
+char* get_time_buffer(){
+ return getTSData(time_now_buffer,TIME_NOW_BUF_SIZE);
+}
+
+char* get_format_log_buffer(){
+ return getTSData(format_log_msg_buffer,FORMAT_LOG_BUF_SIZE);
+}
+#else
+char* get_time_buffer(){
+ static char buf[TIME_NOW_BUF_SIZE];
+ return buf;
+}
+
+char* get_format_log_buffer(){
+ static char buf[FORMAT_LOG_BUF_SIZE];
+ return buf;
+}
+
+#endif
+
+ZooLogLevel logLevel=ZOO_LOG_LEVEL_INFO;
+
+static FILE* logStream=0;
+FILE* getLogStream(){
+ if(logStream==0)
+ logStream=stderr;
+ return logStream;
+}
+
+void zoo_set_log_stream(FILE* stream){
+ logStream=stream;
+}
+
+static const char* time_now(char* now_str){
+ struct timeval tv;
+ struct tm lt;
+ time_t now = 0;
+ size_t len = 0;
+
+ gettimeofday(&tv,0);
+
+ now = tv.tv_sec;
+ localtime_r(&now, <);
+
+ // clone the format used by log4j ISO8601DateFormat
+ // specifically: "yyyy-MM-dd HH:mm:ss,SSS"
+
+ len = strftime(now_str, TIME_NOW_BUF_SIZE,
+ "%Y-%m-%d %H:%M:%S",
+ <);
+
+ len += snprintf(now_str + len,
+ TIME_NOW_BUF_SIZE - len,
+ ",%03d",
+ (int)(tv.tv_usec/1000));
+
+ return now_str;
+}
+
+void log_message(ZooLogLevel curLevel,int line,const char* funcName,
+ const char* message)
+{
+ static const char* dbgLevelStr[]={"ZOO_INVALID","ZOO_ERROR","ZOO_WARN",
+ "ZOO_INFO","ZOO_DEBUG"};
+ static pid_t pid=0;
+#ifdef WIN32
+ char timebuf [TIME_NOW_BUF_SIZE];
+#endif
+ if(pid==0)pid=getpid();
+#ifndef THREADED
+ // pid_t is long on Solaris
+ fprintf(LOGSTREAM, "%s:%ld:%s@%s@%d: %s\n", time_now(get_time_buffer()),(long)pid,
+ dbgLevelStr[curLevel],funcName,line,message);
+#else
+#ifdef WIN32
+ fprintf(LOGSTREAM, "%s:%d(0x%lx):%s@%s@%d: %s\n", time_now(timebuf),pid,
+ (unsigned long int)(pthread_self().thread_id),
+ dbgLevelStr[curLevel],funcName,line,message);
+#else
+ fprintf(LOGSTREAM, "%s:%ld(0x%lx):%s@%s@%d: %s\n", time_now(get_time_buffer()),(long)pid,
+ (unsigned long int)pthread_self(),
+ dbgLevelStr[curLevel],funcName,line,message);
+#endif
+#endif
+ fflush(LOGSTREAM);
+}
+
+const char* format_log_message(const char* format,...)
+{
+ va_list va;
+ char* buf=get_format_log_buffer();
+ if(!buf)
+ return "format_log_message: Unable to allocate memory buffer";
+
+ va_start(va,format);
+ vsnprintf(buf, FORMAT_LOG_BUF_SIZE-1,format,va);
+ va_end(va);
+ return buf;
+}
+
+void zoo_set_debug_level(ZooLogLevel level)
+{
+ if(level==0){
+ // disable logging (unit tests do this)
+ logLevel=(ZooLogLevel)0;
+ return;
+ }
+ if(level<ZOO_LOG_LEVEL_ERROR)level=ZOO_LOG_LEVEL_ERROR;
+ if(level>ZOO_LOG_LEVEL_DEBUG)level=ZOO_LOG_LEVEL_DEBUG;
+ logLevel=level;
+}
+