branch-2.0.4-alpha/hadoop-hdfs-project/hadoop-hdfs/src/contrib/libwebhdfs/src/hdfs_http_client.c - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 #include <stdlib.h>
 #include <string.h>
 #include <curl/curl.h>

 #include "hdfs_http_client.h"

 static pthread_mutex_t curlInitMutex = PTHREAD_MUTEX_INITIALIZER;
 static volatile int curlGlobalInited = 0;

 const char *hdfs_strerror(int errnoval)
 {
     const char *msg = NULL;
     if (errnoval < 0 || errnoval >= sys_nerr) {
         msg = "Invalid Error Code";
     } else if (sys_errlist == NULL) {
         msg = "Unknown Error";
     } else {
         msg = sys_errlist[errnoval];
     }
     return msg;
 }

 int initResponseBuffer(struct ResponseBuffer **buffer)
 {
     struct ResponseBuffer *info = NULL;
     int ret = 0;
     info = calloc(1, sizeof(struct ResponseBuffer));
     if (!info) {
         ret = ENOMEM;
     }
     *buffer = info;
     return ret;
 }

 void freeResponseBuffer(struct ResponseBuffer *buffer)
 {
     if (buffer) {
         if (buffer->content) {
             free(buffer->content);
         }
         free(buffer);
         buffer = NULL;
     }
 }

 void freeResponse(struct Response *resp)
 {
     if (resp) {
         freeResponseBuffer(resp->body);
         freeResponseBuffer(resp->header);
         free(resp);
         resp = NULL;
     }
 }

 /**
  * Callback used by libcurl for allocating local buffer and
  * reading data to local buffer
  */
 static size_t writefunc(void *ptr, size_t size,
                         size_t nmemb, struct ResponseBuffer *rbuffer)
 {
     void *temp = NULL;
     if (size * nmemb < 1) {
         return 0;
     }
     if (!rbuffer) {
         fprintf(stderr,
                 "ERROR: ResponseBuffer is NULL for the callback writefunc.\n");
         return 0;
     }

     if (rbuffer->remaining < size * nmemb) {
         temp = realloc(rbuffer->content, rbuffer->offset + size * nmemb + 1);
         if (temp == NULL) {
             fprintf(stderr, "ERROR: fail to realloc in callback writefunc.\n");
             return 0;
         }
         rbuffer->content = temp;
         rbuffer->remaining = size * nmemb;
     }
     memcpy(rbuffer->content + rbuffer->offset, ptr, size * nmemb);
     rbuffer->offset += size * nmemb;
     (rbuffer->content)[rbuffer->offset] = '\0';
     rbuffer->remaining -= size * nmemb;
     return size * nmemb;
 }

 /**
  * Callback used by libcurl for reading data into buffer provided by user,
  * thus no need to reallocate buffer.
  */
 static size_t writeFuncWithUserBuffer(void *ptr, size_t size,
                                    size_t nmemb, struct ResponseBuffer *rbuffer)
 {
     size_t toCopy = 0;
     if (size * nmemb < 1) {
         return 0;
     }
     if (!rbuffer || !rbuffer->content) {
         fprintf(stderr,
                 "ERROR: buffer to read is NULL for the "
                 "callback writeFuncWithUserBuffer.\n");
         return 0;
     }

     toCopy = rbuffer->remaining < (size * nmemb) ?
                             rbuffer->remaining : (size * nmemb);
     memcpy(rbuffer->content + rbuffer->offset, ptr, toCopy);
     rbuffer->offset += toCopy;
     rbuffer->remaining -= toCopy;
     return toCopy;
 }

 /**
  * Callback used by libcurl for writing data to remote peer
  */
 static size_t readfunc(void *ptr, size_t size, size_t nmemb, void *stream)
 {
     struct webhdfsBuffer *wbuffer = NULL;
     if (size * nmemb < 1) {
         return 0;
     }

     wbuffer = stream;
     pthread_mutex_lock(&wbuffer->writeMutex);
     while (wbuffer->remaining == 0) {
         /*
          * The current remainning bytes to write is 0,
          * check closeFlag to see whether need to finish the transfer.
          * if yes, return 0; else, wait
          */
         if (wbuffer->closeFlag) { // We can close the transfer now
             //For debug
             fprintf(stderr, "CloseFlag is set, ready to close the transfer\n");
             pthread_mutex_unlock(&wbuffer->writeMutex);
             return 0;
         } else {
             // remaining == 0 but closeFlag is not set
             // indicates that user's buffer has been transferred
             pthread_cond_signal(&wbuffer->transfer_finish);
             pthread_cond_wait(&wbuffer->newwrite_or_close,
                                     &wbuffer->writeMutex);
         }
     }

     if (wbuffer->remaining > 0 && !wbuffer->closeFlag) {
         size_t copySize = wbuffer->remaining < size * nmemb ?
                                 wbuffer->remaining : size * nmemb;
         memcpy(ptr, wbuffer->wbuffer + wbuffer->offset, copySize);
         wbuffer->offset += copySize;
         wbuffer->remaining -= copySize;
         pthread_mutex_unlock(&wbuffer->writeMutex);
         return copySize;
     } else {
         fprintf(stderr, "ERROR: webhdfsBuffer's remaining is %ld, "
                 "it should be a positive value!\n", wbuffer->remaining);
         pthread_mutex_unlock(&wbuffer->writeMutex);
         return 0;
     }
 }

 /**
  * Initialize the global libcurl environment
  */
 static void initCurlGlobal()
 {
     if (!curlGlobalInited) {
         pthread_mutex_lock(&curlInitMutex);
         if (!curlGlobalInited) {
             curl_global_init(CURL_GLOBAL_ALL);
             curlGlobalInited = 1;
         }
         pthread_mutex_unlock(&curlInitMutex);
     }
 }

 /**
  * Launch simple commands (commands without file I/O) and return response
  *
  * @param url       Target URL
  * @param method    HTTP method (GET/PUT/POST)
  * @param followloc Whether or not need to set CURLOPT_FOLLOWLOCATION
  * @param response  Response from remote service
  * @return 0 for success and non-zero value to indicate error
  */
 static int launchCmd(const char *url, enum HttpHeader method,
                      enum Redirect followloc, struct Response **response)
 {
     CURL *curl = NULL;
     CURLcode curlCode;
     int ret = 0;
     struct Response *resp = NULL;

     resp = calloc(1, sizeof(struct Response));
     if (!resp) {
         return ENOMEM;
     }
     ret = initResponseBuffer(&(resp->body));
     if (ret) {
         goto done;
     }
     ret = initResponseBuffer(&(resp->header));
     if (ret) {
         goto done;
     }
     initCurlGlobal();
     curl = curl_easy_init();
     if (!curl) {
         ret = ENOMEM;       // curl_easy_init does not return error code,
                             // and most of its errors are caused by malloc()
         fprintf(stderr, "ERROR in curl_easy_init.\n");
         goto done;
     }
     /* Set callback function for reading data from remote service */
     curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writefunc);
     curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
     curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
     curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
     curl_easy_setopt(curl, CURLOPT_URL, url);
     switch(method) {
         case GET:
             break;
         case PUT:
             curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
             break;
         case POST:
             curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
             break;
         case DELETE:
             curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "DELETE");
             break;
         default:
             ret = EINVAL;
             fprintf(stderr, "ERROR: Invalid HTTP method\n");
             goto done;
     }
     if (followloc == YES) {
         curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);
     }
     /* Now run the curl handler */
     curlCode = curl_easy_perform(curl);
     if (curlCode != CURLE_OK) {
         ret = EIO;
         fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
                 url, curlCode, curl_easy_strerror(curlCode));
     }
 done:
     if (curl != NULL) {
         curl_easy_cleanup(curl);
     }
     if (ret) {
         free(resp);
         resp = NULL;
     }
     *response = resp;
     return ret;
 }

 /**
  * Launch the read request. The request is sent to the NameNode and then
  * redirected to corresponding DataNode
  *
  * @param url   The URL for the read request
  * @param resp  The response containing the buffer provided by user
  * @return 0 for success and non-zero value to indicate error
  */
 static int launchReadInternal(const char *url, struct Response* resp)
 {
     CURL *curl;
     CURLcode curlCode;
     int ret = 0;

     if (!resp || !resp->body || !resp->body->content) {
         fprintf(stderr,
                 "ERROR: invalid user-provided buffer!\n");
         return EINVAL;
     }

     initCurlGlobal();
     /* get a curl handle */
     curl = curl_easy_init();
     if (!curl) {
         fprintf(stderr, "ERROR in curl_easy_init.\n");
         return ENOMEM;
     }
     curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeFuncWithUserBuffer);
     curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
     curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
     curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
     curl_easy_setopt(curl, CURLOPT_URL, url);
     curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);

     curlCode = curl_easy_perform(curl);
     if (curlCode != CURLE_OK && curlCode != CURLE_PARTIAL_FILE) {
         ret = EIO;
         fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
                 url, curlCode, curl_easy_strerror(curlCode));
     }

     curl_easy_cleanup(curl);
     return ret;
 }

 /**
  * The function does the write operation by connecting to a DataNode.
  * The function keeps the connection with the DataNode until
  * the closeFlag is set. Whenever the current data has been sent out,
  * the function blocks waiting for further input from user or close.
  *
  * @param url           URL of the remote DataNode
  * @param method        PUT for create and POST for append
  * @param uploadBuffer  Buffer storing user's data to write
  * @param response      Response from remote service
  * @return 0 for success and non-zero value to indicate error
  */
 static int launchWrite(const char *url, enum HttpHeader method,
                        struct webhdfsBuffer *uploadBuffer,
                        struct Response **response)
 {
     CURLcode curlCode;
     struct Response* resp = NULL;
     struct curl_slist *chunk = NULL;
     CURL *curl = NULL;
     int ret = 0;

     if (!uploadBuffer) {
         fprintf(stderr, "ERROR: upload buffer is NULL!\n");
         return EINVAL;
     }

     initCurlGlobal();
     resp = calloc(1, sizeof(struct Response));
     if (!resp) {
         return ENOMEM;
     }
     ret = initResponseBuffer(&(resp->body));
     if (ret) {
         goto done;
     }
     ret = initResponseBuffer(&(resp->header));
     if (ret) {
         goto done;
     }

     // Connect to the datanode in order to create the lease in the namenode
     curl = curl_easy_init();
     if (!curl) {
         fprintf(stderr, "ERROR: failed to initialize the curl handle.\n");
         return ENOMEM;
     }
     curl_easy_setopt(curl, CURLOPT_URL, url);

     curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writefunc);
     curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
     curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
     curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
     curl_easy_setopt(curl, CURLOPT_READFUNCTION, readfunc);
     curl_easy_setopt(curl, CURLOPT_READDATA, uploadBuffer);
     curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);

     chunk = curl_slist_append(chunk, "Transfer-Encoding: chunked");
     curl_easy_setopt(curl, CURLOPT_HTTPHEADER, chunk);
     chunk = curl_slist_append(chunk, "Expect:");
     curl_easy_setopt(curl, CURLOPT_HTTPHEADER, chunk);

     switch(method) {
         case PUT:
             curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
             break;
         case POST:
             curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
             break;
         default:
             ret = EINVAL;
             fprintf(stderr, "ERROR: Invalid HTTP method\n");
             goto done;
     }
     curlCode = curl_easy_perform(curl);
     if (curlCode != CURLE_OK) {
         ret = EIO;
         fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
                 url, curlCode, curl_easy_strerror(curlCode));
     }

 done:
     if (chunk != NULL) {
         curl_slist_free_all(chunk);
     }
     if (curl != NULL) {
         curl_easy_cleanup(curl);
     }
     if (ret) {
         free(resp);
         resp = NULL;
     }
     *response = resp;
     return ret;
 }

 int launchMKDIR(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchRENAME(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchGFS(const char *url, struct Response **resp)
 {
     return launchCmd(url, GET, NO, resp);
 }

 int launchLS(const char *url, struct Response **resp)
 {
     return launchCmd(url, GET, NO, resp);
 }

 int launchCHMOD(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchCHOWN(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchDELETE(const char *url, struct Response **resp)
 {
     return launchCmd(url, DELETE, NO, resp);
 }

 int launchOPEN(const char *url, struct Response* resp)
 {
     return launchReadInternal(url, resp);
 }

 int launchUTIMES(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchNnWRITE(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }

 int launchNnAPPEND(const char *url, struct Response **resp)
 {
     return launchCmd(url, POST, NO, resp);
 }

 int launchDnWRITE(const char *url, struct webhdfsBuffer *buffer,
                                struct Response **resp)
 {
     return launchWrite(url, PUT, buffer, resp);
 }

 int launchDnAPPEND(const char *url, struct webhdfsBuffer *buffer,
                                 struct Response **resp)
 {
     return launchWrite(url, POST, buffer, resp);
 }

 int launchSETREPLICATION(const char *url, struct Response **resp)
 {
     return launchCmd(url, PUT, NO, resp);
 }
	/**
	* 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 <stdlib.h>
	#include <string.h>
	#include <curl/curl.h>

	#include "hdfs_http_client.h"

	static pthread_mutex_t curlInitMutex = PTHREAD_MUTEX_INITIALIZER;
	static volatile int curlGlobalInited = 0;

	const char *hdfs_strerror(int errnoval)
	{
	const char *msg = NULL;
	if (errnoval < 0 \|\| errnoval >= sys_nerr) {
	msg = "Invalid Error Code";
	} else if (sys_errlist == NULL) {
	msg = "Unknown Error";
	} else {
	msg = sys_errlist[errnoval];
	}
	return msg;
	}

	int initResponseBuffer(struct ResponseBuffer **buffer)
	{
	struct ResponseBuffer *info = NULL;
	int ret = 0;
	info = calloc(1, sizeof(struct ResponseBuffer));
	if (!info) {
	ret = ENOMEM;
	}
	*buffer = info;
	return ret;
	}

	void freeResponseBuffer(struct ResponseBuffer *buffer)
	{
	if (buffer) {
	if (buffer->content) {
	free(buffer->content);
	}
	free(buffer);
	buffer = NULL;
	}
	}

	void freeResponse(struct Response *resp)
	{
	if (resp) {
	freeResponseBuffer(resp->body);
	freeResponseBuffer(resp->header);
	free(resp);
	resp = NULL;
	}
	}

	/**
	* Callback used by libcurl for allocating local buffer and
	* reading data to local buffer
	*/
	static size_t writefunc(void *ptr, size_t size,
	size_t nmemb, struct ResponseBuffer *rbuffer)
	{
	void *temp = NULL;
	if (size * nmemb < 1) {
	return 0;
	}
	if (!rbuffer) {
	fprintf(stderr,
	"ERROR: ResponseBuffer is NULL for the callback writefunc.\n");
	return 0;
	}

	if (rbuffer->remaining < size * nmemb) {
	temp = realloc(rbuffer->content, rbuffer->offset + size * nmemb + 1);
	if (temp == NULL) {
	fprintf(stderr, "ERROR: fail to realloc in callback writefunc.\n");
	return 0;
	}
	rbuffer->content = temp;
	rbuffer->remaining = size * nmemb;
	}
	memcpy(rbuffer->content + rbuffer->offset, ptr, size * nmemb);
	rbuffer->offset += size * nmemb;
	(rbuffer->content)[rbuffer->offset] = '\0';
	rbuffer->remaining -= size * nmemb;
	return size * nmemb;
	}

	/**
	* Callback used by libcurl for reading data into buffer provided by user,
	* thus no need to reallocate buffer.
	*/
	static size_t writeFuncWithUserBuffer(void *ptr, size_t size,
	size_t nmemb, struct ResponseBuffer *rbuffer)
	{
	size_t toCopy = 0;
	if (size * nmemb < 1) {
	return 0;
	}
	if (!rbuffer \|\| !rbuffer->content) {
	fprintf(stderr,
	"ERROR: buffer to read is NULL for the "
	"callback writeFuncWithUserBuffer.\n");
	return 0;
	}

	toCopy = rbuffer->remaining < (size * nmemb) ?
	rbuffer->remaining : (size * nmemb);
	memcpy(rbuffer->content + rbuffer->offset, ptr, toCopy);
	rbuffer->offset += toCopy;
	rbuffer->remaining -= toCopy;
	return toCopy;
	}

	/**
	* Callback used by libcurl for writing data to remote peer
	*/
	static size_t readfunc(void ptr, size_t size, size_t nmemb, void stream)
	{
	struct webhdfsBuffer *wbuffer = NULL;
	if (size * nmemb < 1) {
	return 0;
	}

	wbuffer = stream;
	pthread_mutex_lock(&wbuffer->writeMutex);
	while (wbuffer->remaining == 0) {
	/*
	* The current remainning bytes to write is 0,
	* check closeFlag to see whether need to finish the transfer.
	* if yes, return 0; else, wait
	*/
	if (wbuffer->closeFlag) { // We can close the transfer now
	//For debug
	fprintf(stderr, "CloseFlag is set, ready to close the transfer\n");
	pthread_mutex_unlock(&wbuffer->writeMutex);
	return 0;
	} else {
	// remaining == 0 but closeFlag is not set
	// indicates that user's buffer has been transferred
	pthread_cond_signal(&wbuffer->transfer_finish);
	pthread_cond_wait(&wbuffer->newwrite_or_close,
	&wbuffer->writeMutex);
	}
	}

	if (wbuffer->remaining > 0 && !wbuffer->closeFlag) {
	size_t copySize = wbuffer->remaining < size * nmemb ?
	wbuffer->remaining : size * nmemb;
	memcpy(ptr, wbuffer->wbuffer + wbuffer->offset, copySize);
	wbuffer->offset += copySize;
	wbuffer->remaining -= copySize;
	pthread_mutex_unlock(&wbuffer->writeMutex);
	return copySize;
	} else {
	fprintf(stderr, "ERROR: webhdfsBuffer's remaining is %ld, "
	"it should be a positive value!\n", wbuffer->remaining);
	pthread_mutex_unlock(&wbuffer->writeMutex);
	return 0;
	}
	}

	/**
	* Initialize the global libcurl environment
	*/
	static void initCurlGlobal()
	{
	if (!curlGlobalInited) {
	pthread_mutex_lock(&curlInitMutex);
	if (!curlGlobalInited) {
	curl_global_init(CURL_GLOBAL_ALL);
	curlGlobalInited = 1;
	}
	pthread_mutex_unlock(&curlInitMutex);
	}
	}

	/**
	* Launch simple commands (commands without file I/O) and return response
	*
	* @param url Target URL
	* @param method HTTP method (GET/PUT/POST)
	* @param followloc Whether or not need to set CURLOPT_FOLLOWLOCATION
	* @param response Response from remote service
	* @return 0 for success and non-zero value to indicate error
	*/
	static int launchCmd(const char *url, enum HttpHeader method,
	enum Redirect followloc, struct Response **response)
	{
	CURL *curl = NULL;
	CURLcode curlCode;
	int ret = 0;
	struct Response *resp = NULL;

	resp = calloc(1, sizeof(struct Response));
	if (!resp) {
	return ENOMEM;
	}
	ret = initResponseBuffer(&(resp->body));
	if (ret) {
	goto done;
	}
	ret = initResponseBuffer(&(resp->header));
	if (ret) {
	goto done;
	}
	initCurlGlobal();
	curl = curl_easy_init();
	if (!curl) {
	ret = ENOMEM; // curl_easy_init does not return error code,
	// and most of its errors are caused by malloc()
	fprintf(stderr, "ERROR in curl_easy_init.\n");
	goto done;
	}
	/* Set callback function for reading data from remote service */
	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writefunc);
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
	curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
	curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
	curl_easy_setopt(curl, CURLOPT_URL, url);
	switch(method) {
	case GET:
	break;
	case PUT:
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
	break;
	case POST:
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
	break;
	case DELETE:
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "DELETE");
	break;
	default:
	ret = EINVAL;
	fprintf(stderr, "ERROR: Invalid HTTP method\n");
	goto done;
	}
	if (followloc == YES) {
	curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);
	}
	/* Now run the curl handler */
	curlCode = curl_easy_perform(curl);
	if (curlCode != CURLE_OK) {
	ret = EIO;
	fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
	url, curlCode, curl_easy_strerror(curlCode));
	}
	done:
	if (curl != NULL) {
	curl_easy_cleanup(curl);
	}
	if (ret) {
	free(resp);
	resp = NULL;
	}
	*response = resp;
	return ret;
	}

	/**
	* Launch the read request. The request is sent to the NameNode and then
	* redirected to corresponding DataNode
	*
	* @param url The URL for the read request
	* @param resp The response containing the buffer provided by user
	* @return 0 for success and non-zero value to indicate error
	*/
	static int launchReadInternal(const char url, struct Response resp)
	{
	CURL *curl;
	CURLcode curlCode;
	int ret = 0;

	if (!resp \|\| !resp->body \|\| !resp->body->content) {
	fprintf(stderr,
	"ERROR: invalid user-provided buffer!\n");
	return EINVAL;
	}

	initCurlGlobal();
	/* get a curl handle */
	curl = curl_easy_init();
	if (!curl) {
	fprintf(stderr, "ERROR in curl_easy_init.\n");
	return ENOMEM;
	}
	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeFuncWithUserBuffer);
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
	curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
	curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
	curl_easy_setopt(curl, CURLOPT_URL, url);
	curl_easy_setopt(curl, CURLOPT_FOLLOWLOCATION, 1);

	curlCode = curl_easy_perform(curl);
	if (curlCode != CURLE_OK && curlCode != CURLE_PARTIAL_FILE) {
	ret = EIO;
	fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
	url, curlCode, curl_easy_strerror(curlCode));
	}

	curl_easy_cleanup(curl);
	return ret;
	}

	/**
	* The function does the write operation by connecting to a DataNode.
	* The function keeps the connection with the DataNode until
	* the closeFlag is set. Whenever the current data has been sent out,
	* the function blocks waiting for further input from user or close.
	*
	* @param url URL of the remote DataNode
	* @param method PUT for create and POST for append
	* @param uploadBuffer Buffer storing user's data to write
	* @param response Response from remote service
	* @return 0 for success and non-zero value to indicate error
	*/
	static int launchWrite(const char *url, enum HttpHeader method,
	struct webhdfsBuffer *uploadBuffer,
	struct Response **response)
	{
	CURLcode curlCode;
	struct Response* resp = NULL;
	struct curl_slist *chunk = NULL;
	CURL *curl = NULL;
	int ret = 0;

	if (!uploadBuffer) {
	fprintf(stderr, "ERROR: upload buffer is NULL!\n");
	return EINVAL;
	}

	initCurlGlobal();
	resp = calloc(1, sizeof(struct Response));
	if (!resp) {
	return ENOMEM;
	}
	ret = initResponseBuffer(&(resp->body));
	if (ret) {
	goto done;
	}
	ret = initResponseBuffer(&(resp->header));
	if (ret) {
	goto done;
	}

	// Connect to the datanode in order to create the lease in the namenode
	curl = curl_easy_init();
	if (!curl) {
	fprintf(stderr, "ERROR: failed to initialize the curl handle.\n");
	return ENOMEM;
	}
	curl_easy_setopt(curl, CURLOPT_URL, url);

	curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writefunc);
	curl_easy_setopt(curl, CURLOPT_WRITEDATA, resp->body);
	curl_easy_setopt(curl, CURLOPT_HEADERFUNCTION, writefunc);
	curl_easy_setopt(curl, CURLOPT_WRITEHEADER, resp->header);
	curl_easy_setopt(curl, CURLOPT_READFUNCTION, readfunc);
	curl_easy_setopt(curl, CURLOPT_READDATA, uploadBuffer);
	curl_easy_setopt(curl, CURLOPT_UPLOAD, 1L);

	chunk = curl_slist_append(chunk, "Transfer-Encoding: chunked");
	curl_easy_setopt(curl, CURLOPT_HTTPHEADER, chunk);
	chunk = curl_slist_append(chunk, "Expect:");
	curl_easy_setopt(curl, CURLOPT_HTTPHEADER, chunk);

	switch(method) {
	case PUT:
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PUT");
	break;
	case POST:
	curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "POST");
	break;
	default:
	ret = EINVAL;
	fprintf(stderr, "ERROR: Invalid HTTP method\n");
	goto done;
	}
	curlCode = curl_easy_perform(curl);
	if (curlCode != CURLE_OK) {
	ret = EIO;
	fprintf(stderr, "ERROR: preform the URL %s failed, <%d>: %s\n",
	url, curlCode, curl_easy_strerror(curlCode));
	}

	done:
	if (chunk != NULL) {
	curl_slist_free_all(chunk);
	}
	if (curl != NULL) {
	curl_easy_cleanup(curl);
	}
	if (ret) {
	free(resp);
	resp = NULL;
	}
	*response = resp;
	return ret;
	}

	int launchMKDIR(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchRENAME(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchGFS(const char url, struct Response *resp)
	{
	return launchCmd(url, GET, NO, resp);
	}

	int launchLS(const char url, struct Response *resp)
	{
	return launchCmd(url, GET, NO, resp);
	}

	int launchCHMOD(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchCHOWN(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchDELETE(const char url, struct Response *resp)
	{
	return launchCmd(url, DELETE, NO, resp);
	}

	int launchOPEN(const char url, struct Response resp)
	{
	return launchReadInternal(url, resp);
	}

	int launchUTIMES(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchNnWRITE(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}

	int launchNnAPPEND(const char url, struct Response *resp)
	{
	return launchCmd(url, POST, NO, resp);
	}

	int launchDnWRITE(const char url, struct webhdfsBuffer buffer,
	struct Response **resp)
	{
	return launchWrite(url, PUT, buffer, resp);
	}

	int launchDnAPPEND(const char url, struct webhdfsBuffer buffer,
	struct Response **resp)
	{
	return launchWrite(url, POST, buffer, resp);
	}

	int launchSETREPLICATION(const char url, struct Response *resp)
	{
	return launchCmd(url, PUT, NO, resp);
	}