Handle DP operation based on DP_OPEN_STATE

Change-Id: Ie0d99153e03b57543b3ea632f0140b730c1f8eb9

diff --git a/plugins/libwebsockets/libwebsockets-plugin.cpp b/plugins/libwebsockets/libwebsockets-plugin.cpp
index 942d99228dac7c27cf2456e11ab7db1db49c0afc..dcfa0851ff9c76d45512f65c973a7a7b7c6926cc 100755 (executable)
--- a/plugins/libwebsockets/libwebsockets-plugin.cpp
+++ b/plugins/libwebsockets/libwebsockets-plugin.cpp
@@ -798,7 +798,8 @@ static void _request_write(websocket_s *ws)
                return;
 
        VINE_LOGI("Request write callback for %p", ws);
-       lws_callback_on_writable(ws->wsi);
+       if (lws_callback_on_writable(ws->wsi) < 1)
+               VINE_LOGI("Failed to request a writable callback.");
 }
 
 static int websocket_write(vine_dp_plugin_h handle, unsigned char *buf, size_t len)

diff --git a/src/include/vine-data-path-state.h b/src/include/vine-data-path-state.h
index f68f3a03ef7dc67022dacf90f139342f854cf3b8..4147ecbab711840797d58670a585162f4fa6d09d 100644 (file)
--- a/src/include/vine-data-path-state.h
+++ b/src/include/vine-data-path-state.h
@@ -249,5 +249,6 @@ void vine_dp_state_set_established_notifier(established_notifier notifier);
 void vine_dp_state_set_data_received_notifier(data_received_notifier notifier);
 
 VineDataPathState *vine_get_default_state(vine_data_path_h dp, vine_dp_plugin_h plugin);
+void start_default_state(vine_data_path_h dp, vine_dp_plugin_h plugin, VineDataPathState *state);
 void start_connected_state(vine_data_path_h dp, vine_dp_plugin_h plugin, VineDataPathState *state);
 void start_accepted_state(vine_data_path_h dp, vine_dp_plugin_h plugin, VineDataPathState *state);

diff --git a/src/include/vine-dp.h b/src/include/vine-dp.h
index 821ffff71a0c080f2cd644d91c8708372edce1e1..5e1ac7b75469216a9b4242884ce96940702d6b93 100644 (file)
--- a/src/include/vine-dp.h
+++ b/src/include/vine-dp.h
@@ -30,6 +30,12 @@ namespace vine {
 #define VINE_DP_DEFAULT_CONNECTIONS_NUM 30
 #define VINE_DP_MAX_CONNECTIONS_NUM 200
 
+enum {
+       VINE_DP_OPEN_STATE_NONE = 0,
+       VINE_DP_OPEN_STATE_WAIT,
+       VINE_DP_OPEN_STATE_DONE,
+};
+
 class DataPath
 {
 public:
@@ -67,6 +73,7 @@ protected:
        void *mEventFd;
        void *mSecurity;
        std::string mIfaceName;
+       int mOpenState;
 
        vine_dp_opened_cb mOpenedCb;
        void *mOpenedCbData;
@@ -138,13 +145,6 @@ private:
 };
 
 typedef std::map<std::string, vine_data_path_h> DPMap;
-
-enum {
-       VINE_DP_PUBSUB_OPEN_STATE_NONE = 0,
-       VINE_DP_PUBSUB_OPEN_STATE_WAIT,
-       VINE_DP_PUBSUB_OPEN_STATE_DONE,
-};
-
 enum {
        VINE_DP_PUBSUB_SD_STATE_NONE = 1 << 0,
        VINE_DP_PUBSUB_SD_STATE_PUBLISH = 1 << 1,
@@ -223,7 +223,6 @@ private:
        int compare_ip_priority(const char *peer_ip);
 
        int mInitDiscNum;
-       int mOpenState;
        int mSdPubSubState;
        static void _open_timer(void *user_data);
        vine::VineTimer mOpenTimer;

diff --git a/src/vine-data-path-state.cpp b/src/vine-data-path-state.cpp
index 1254ca3425fa61abe677317a9034b81a68f8eb79..eb64421137aa19985a97ed6da51345a132ea45be 100644 (file)
--- a/src/vine-data-path-state.cpp
+++ b/src/vine-data-path-state.cpp
@@ -753,6 +753,13 @@ VineDataPathState *vine_get_default_state(vine_data_path_h dp, vine_dp_plugin_h
        return new VineDefaultState(dp, plugin);
 }
 
+void start_default_state(vine_data_path_h dp, vine_dp_plugin_h plugin,
+               VineDataPathState *state)
+{
+       VINE_LOGD("Default");
+       state_transition(dp, state, new VineDefaultState(dp, plugin));
+}
+
 void start_connected_state(vine_data_path_h dp, vine_dp_plugin_h plugin,
        VineDataPathState *state)
 {

diff --git a/src/vine-data-path.cpp b/src/vine-data-path.cpp
index 3a48609aaff3ad230d1edf394fc74cd0bb36d0df..6971529e8b8eb022b493df457a74cd41c4d515e6 100755 (executable)
--- a/src/vine-data-path.cpp
+++ b/src/vine-data-path.cpp
@@ -367,6 +367,7 @@ static void __invoke_terminated_user_cb(void *event, void *user_data)
 
        VINE_LOGD("user callback is invoked by event queue.");
 
+       start_default_state(dp, dp->plugin_handle, dp->state);
        if (dp && dp->terminated_cb)
                dp->terminated_cb(dp, dp->terminated_cb_data);
 }
@@ -673,6 +674,10 @@ int vine_data_path_open(vine_address_family_e addr_family, int port, const char
                return __convert_data_path_error_to_vine_error((vine_data_path_error)ret);
        }
 
+       if (*opened_datapath) {
+               VINE_LOGI("Destroy a stale datapath.");
+               _vine_data_path_destroy(*opened_datapath);
+       }
        *opened_datapath = dp;
 
        return VINE_ERROR_NONE;
@@ -713,6 +718,11 @@ int vine_data_path_connect(vine_address_family_e addr_family,
                        ip, port, iface_name, ssl);
        _destroy_security_info(&ssl);
 
+       if (*connected_datapath) {
+               VINE_LOGI("Destroy a stale datapath.");
+               _vine_data_path_destroy(*connected_datapath);
+       }
+
        if (connected_datapath)
                *connected_datapath = dp;
 

diff --git a/src/vine-dp.cpp b/src/vine-dp.cpp
index 5437d3e8f691ec3b2cd164fa616e6ad3971a84dc..a387e1c1d4b0a5c5b715baf94e77655b22703218 100644 (file)
--- a/src/vine-dp.cpp
+++ b/src/vine-dp.cpp
@@ -33,6 +33,17 @@
 
 using namespace vine;
 
+#define RET_ERR_IF_DP_OPEN_STATE_ISNT_IDLE(state) \
+       do { \
+        if (state == VINE_DP_OPEN_STATE_WAIT) { \
+                       VINE_LOGE("Ignore duplicate request."); \
+            return VINE_ERROR_NOW_IN_PROGRESS; \
+        } else if (state == VINE_DP_OPEN_STATE_DONE) { \
+                       VINE_LOGE("Already opened."); \
+                       return VINE_ERROR_INVALID_OPERATION; \
+               } \
+    } while (0)
+
 extern vine_dp_plugin_fn g_dp_plugin_fn;
 
 static bool _check_if_valid_ip(vine_address_family_e addr_family, const char *ip)
@@ -181,9 +192,9 @@ static void _pubsub_accepted_cb(vine_data_path_h datapath, void *user_data)
        _vine_data_path_set_received_cb(datapath, _pubsub_received_cb, user_data);
        vine_data_path_set_terminated_cb(datapath, _pubsub_terminated_cb, user_data);
 
-       if (dp->get_open_state() == VINE_DP_PUBSUB_OPEN_STATE_WAIT
+       if (dp->get_open_state() == VINE_DP_OPEN_STATE_WAIT
                        && dp->decrease_init_disc_num() <= 0) {
-               dp->set_open_state(VINE_DP_PUBSUB_OPEN_STATE_DONE);
+               dp->set_open_state(VINE_DP_OPEN_STATE_DONE);
                static_cast<DataPath *>(user_data)->invoke_opened_cb(VINE_ERROR_NONE);
        }
 }
@@ -215,9 +226,9 @@ static void _pubsub_connected_cb(vine_data_path_h datapath, int result, void *us
        _vine_data_path_set_received_cb(datapath, _pubsub_received_cb, user_data);
        vine_data_path_set_terminated_cb(datapath, _pubsub_terminated_cb, user_data);
 
-       if (dp->get_open_state() == VINE_DP_PUBSUB_OPEN_STATE_WAIT
+       if (dp->get_open_state() == VINE_DP_OPEN_STATE_WAIT
                        && dp->decrease_init_disc_num() <= 0) {
-               dp->set_open_state(VINE_DP_PUBSUB_OPEN_STATE_DONE);
+               dp->set_open_state(VINE_DP_OPEN_STATE_DONE);
                static_cast<DataPath *>(user_data)->invoke_opened_cb(VINE_ERROR_NONE);
        }
 }
@@ -333,7 +344,7 @@ static void _service_discovered_cb(vine_disc_h disc, bool available,
        }
 
        DPPubSub *dp = static_cast<DPPubSub *>(user_data);
-       if (more_coming && dp->get_open_state() == VINE_DP_PUBSUB_OPEN_STATE_WAIT) {
+       if (more_coming && dp->get_open_state() == VINE_DP_OPEN_STATE_WAIT) {
                VINE_LOGD("At least one more result is exist.");
                dp->increase_init_disc_num();
        }
@@ -387,6 +398,7 @@ void DataPath::invoke_opened_cb(int result)
        // called only once.
        mOpenedCb = NULL;
        mOpenedCbData = NULL;
+       mOpenState = VINE_DP_OPEN_STATE_DONE;
 }
 
 int DataPath::set_received_cb(vine_dp_received_cb callback, void *user_data)
@@ -425,6 +437,7 @@ int DataPath::unset_terminated_cb()
 
 void DataPath::invoke_terminated_cb()
 {
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        if (mTerminatedCb)
                mTerminatedCb(static_cast<void *>(this), mTerminatedCbData);
 }
@@ -436,6 +449,7 @@ DPServer::DPServer(void *event_fd)
        mSecurity = NULL;
        mAddrFamily = VINE_ADDRESS_FAMILY_DEFAULT;
        mIfaceName = "";
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        mDataPath = NULL;
        mListenPort = 0;
        mMaxConnNum = VINE_DP_DEFAULT_CONNECTIONS_NUM;
@@ -513,19 +527,27 @@ void DPServer::invoke_accepted_cb(vine_dp_h dp)
 
 int DPServer::open(vine_dp_opened_cb callback, void *user_data)
 {
-       const char *iface_name = mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL;
+       RET_ERR_IF_DP_OPEN_STATE_ISNT_IDLE(mOpenState);
 
        mOpenedCb = callback;
        mOpenedCbData = user_data;
+       mOpenState = VINE_DP_OPEN_STATE_WAIT;
 
-       return vine_data_path_open(mAddrFamily, mListenPort, iface_name, mMaxConnNum, mSecurity,
+       int ret = vine_data_path_open(mAddrFamily, mListenPort,
+                       mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL,
+                       mMaxConnNum, mSecurity,
                        _opened_cb, static_cast<void *>(this),
                        _accepted_cb, static_cast<void *>(this),
                        &mDataPath, mEventFd);
+       if (ret != VINE_ERROR_NONE)
+               mOpenState = VINE_DP_OPEN_STATE_NONE;
+
+       return ret;
 }
 
 void DPServer::close()
 {
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        _vine_data_path_close(mDataPath);
 }
 
@@ -562,6 +584,7 @@ DPClient::DPClient(void *event_fd, void *datapath)
        VINE_LOGD("DPClient[%p] is created with datapath[%p]", this, datapath);
        mEventFd = event_fd;
        mSecurity = NULL;
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        mDataPath = datapath;
        mAddrFamily = VINE_ADDRESS_FAMILY_IPV4;
        mServerIp = "";
@@ -624,18 +647,24 @@ int DPClient::unset_accepted_cb()
 
 int DPClient::open(vine_dp_opened_cb callback, void *user_data)
 {
-       const char *iface_name = mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL;
+       RET_ERR_IF_DP_OPEN_STATE_ISNT_IDLE(mOpenState);
 
        mOpenedCb = callback;
        mOpenedCbData = user_data;
+       mOpenState = VINE_DP_OPEN_STATE_WAIT;
 
-       return vine_data_path_connect(mAddrFamily, mServerIp.c_str(), mServerPort,
-                       iface_name, mSecurity, NULL,
+       int ret = vine_data_path_connect(mAddrFamily, mServerIp.c_str(), mServerPort,
+                       mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL,
+                       mSecurity, NULL,
                        _connected_cb, static_cast<void *>(this), &mDataPath, mEventFd);
+       if (ret != VINE_ERROR_NONE)
+               mOpenState = VINE_DP_OPEN_STATE_NONE;
+       return ret;
 }
 
 void DPClient::close()
 {
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        _vine_data_path_close(mDataPath);
 }
 
@@ -655,6 +684,7 @@ DPPubSub::DPPubSub(void *event_fd)
        mEventFd = event_fd;
        mSecurity = NULL;
        mIfaceName = "";
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        mReceivedCb = NULL;
        mReceivedCbData = NULL;
        mOpenedCb = NULL;
@@ -668,7 +698,6 @@ DPPubSub::DPPubSub(void *event_fd)
        mSdPub = NULL;
        mServiceName = "";
        mInitDiscNum = 1;
-       mOpenState = VINE_DP_PUBSUB_OPEN_STATE_NONE;
        mSdPubSubState = VINE_DP_PUBSUB_SD_STATE_NONE;
 
        // Network Info
@@ -734,7 +763,7 @@ int DPPubSub::unset_accepted_cb()
 
 int DPPubSub::connect(const char *ip, int port)
 {
-       vine_data_path_h datapath;
+       vine_data_path_h datapath = NULL;
        const char *iface_name = mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL;
        int ret = vine_data_path_connect(mAddrFamily, ip, port, iface_name,
                        mSecurity, mServiceName.c_str(),
@@ -827,28 +856,24 @@ int DPPubSub::create_rank()
 
 int DPPubSub::open(vine_dp_opened_cb callback, void *user_data)
 {
-       if (mOpenState == VINE_DP_PUBSUB_OPEN_STATE_WAIT) {
-               VINE_LOGE("Ignore duplicate request.");
-               return VINE_ERROR_NOW_IN_PROGRESS;
-       } else if (mOpenState == VINE_DP_PUBSUB_OPEN_STATE_DONE) {
-               VINE_LOGE("Already opened.");
-               return VINE_ERROR_INVALID_OPERATION;
-       }
+       RET_ERR_IF_DP_OPEN_STATE_ISNT_IDLE(mOpenState);
 
-       mOpenState = VINE_DP_PUBSUB_OPEN_STATE_WAIT;
+       mOpenState = VINE_DP_OPEN_STATE_WAIT;
        mOpenTimer.start(VINE_DP_PUBSUB_OPEN_TIMEOUT_MS, _open_timer, static_cast<void *>(this));
 
        mOpenedCb = callback;
        mOpenedCbData = user_data;
 
-       const char *iface_name = mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL;
-       int ret = vine_data_path_open(mAddrFamily, mListenPort, iface_name, mMaxConnNum, mSecurity,
+       int ret = vine_data_path_open(mAddrFamily, mListenPort,
+                       mIfaceName.size() > 0 ? mIfaceName.c_str() : NULL,
+                       mMaxConnNum, mSecurity,
                        _pubsub_opened_cb, static_cast<void *>(this),
                        _pubsub_accepted_cb, static_cast<void *>(this),
                        &mServerDataPath, mEventFd);
        if (ret != VINE_ERROR_NONE) {
                mOpenedCb = NULL;
                mOpenedCbData = NULL;
+               mOpenState = VINE_DP_OPEN_STATE_NONE;
                mOpenTimer.stop();
                return ret;
        }
@@ -872,7 +897,7 @@ void DPPubSub::close()
        _vine_data_path_destroy(mServerDataPath);
        mServerDataPath = NULL;
 
-       mOpenState = VINE_DP_PUBSUB_OPEN_STATE_NONE;
+       mOpenState = VINE_DP_OPEN_STATE_NONE;
        mSdPubSubState = VINE_DP_PUBSUB_SD_STATE_NONE;
 }
 
@@ -1071,7 +1096,7 @@ void DPPubSub::_open_timer(void *user_data)
        int sd_state = dp->get_sd_pubsub_state();
        if ((sd_state & VINE_DP_PUBSUB_SD_STATE_PUBLISH)
                        && (sd_state & VINE_DP_PUBSUB_SD_STATE_SUBSCRIBE)) {
-               dp->set_open_state(VINE_DP_PUBSUB_OPEN_STATE_DONE);
+               dp->set_open_state(VINE_DP_OPEN_STATE_DONE);
                static_cast<DataPath *>(user_data)->invoke_opened_cb(VINE_ERROR_NONE);
                return;
        }

diff --git a/tool/tool_run.cpp b/tool/tool_run.cpp
index 633929d6c3e074b627ad234d90e129a32193205b..f6a44e78a134e36a8f0d05f0733bf858e96759a7 100644 (file)
--- a/tool/tool_run.cpp
+++ b/tool/tool_run.cpp
@@ -272,10 +272,16 @@ static int _send_message_from_file(vine_dp_h dp)
        fclose(file);
 
        ret = vine_dp_send(dp, buf, count);
+       const char *ret_str;
+       if (ret == VINE_ERROR_NONE) {
+               printf("Sent total %zd bytes.\n", count);
+               ret_str = "Succeeded";
+       } else {
+               ret_str = "Failed";
+       }
+       printf("%s to send a message.\n", ret_str);
        free(buf);
 
-       printf("Sent total %zd bytes.\n", count);
-
        return ret;
 }
 
@@ -289,6 +295,8 @@ static int _send_message(vine_dp_h dp)
        memcpy(buf, vine_configs.msg, len);
 
        int ret = vine_dp_send(dp, buf, len);
+       printf("%s to send a message.\n", ret == VINE_ERROR_NONE ? "Succeeded" : "Failed");
+
        _start_message_timer(dp, vine_configs.interval);
        return ret;
 }