add HAL socket

this patch contains the following changes:
1. added header and implemenation files of socket util apis
in libgps.utils; 2. updated XtraSystemStatusObserver to use
the new apis; 3. added HAL socket and new thread to listen to it.

Change-Id: If1f6b4b4d6ea2d03640f68e96f0286300404f42b
CRs-fixed: 2108635

6 files changed, 322 insertions, 69 deletions

diff --git a/gnss/XtraSystemStatusObserver.cpp b/gnss/XtraSystemStatusObserver.cpp
index acce6b5..3c08cf3 100644
--- a/gnss/XtraSystemStatusObserver.cpp
+++ b/gnss/XtraSystemStatusObserver.cpp
@@ -51,14 +51,16 @@
 
 using namespace loc_core;
 
-#define XTRA_HAL_SOCKET_NAME "/data/vendor/location/xtra/socket_hal_xtra"
+#ifdef LOG_TAG
+#undef LOG_TAG
+#endif
+#define LOG_TAG "LocSvc_XSSO"
 
 bool XtraSystemStatusObserver::updateLockStatus(uint32_t lock) {
     stringstream ss;
     ss <<  "gpslock";
     ss << " " << lock;
-    ss << "\n"; // append seperator
-    return ( sendEvent(ss) );
+    return ( send(LOC_IPC_XTRA, ss.str()) );
 }
 
 bool XtraSystemStatusObserver::updateConnectionStatus(bool connected, int32_t type) {
@@ -66,79 +68,28 @@ bool XtraSystemStatusObserver::updateConnectionStatus(bool connected, int32_t ty
     ss <<  "connection";
     ss << " " << (connected ? "1" : "0");
     ss << " " << type;
-    ss << "\n"; // append seperator
-    return ( sendEvent(ss) );
+    return ( send(LOC_IPC_XTRA, ss.str()) );
 }
 bool XtraSystemStatusObserver::updateTac(const string& tac) {
     stringstream ss;
     ss <<  "tac";
     ss << " " << tac.c_str();
-    ss << "\n"; // append seperator
-    return ( sendEvent(ss) );
+    return ( send(LOC_IPC_XTRA, ss.str()) );
 }
 
 bool XtraSystemStatusObserver::updateMccMnc(const string& mccmnc) {
     stringstream ss;
     ss <<  "mncmcc";
     ss << " " << mccmnc.c_str();
-    ss << "\n"; // append seperator
-    return ( sendEvent(ss) );
+    return ( send(LOC_IPC_XTRA, ss.str()) );
 }
 
-bool XtraSystemStatusObserver::sendEvent(const stringstream& event) {
-    int socketFd = createSocket();
-    if (socketFd < 0) {
-        LOC_LOGe("XTRA unreachable. sending failed.");
-        return false;
-    }
-
-    const string& data = event.str();
-    int remain = data.length();
-    ssize_t sent = 0;
-    while (remain > 0 &&
-          (sent = ::send(socketFd, data.c_str() + (data.length() - remain),
-                       remain, MSG_NOSIGNAL)) > 0) {
-        remain -= sent;
-    }
-
-    if (sent < 0) {
-        LOC_LOGe("sending error. reason:%s", strerror(errno));
-    }
-
-    closeSocket(socketFd);
-
-    return (remain == 0);
-}
-
-
-int XtraSystemStatusObserver::createSocket() {
-    int socketFd = -1;
+void XtraSystemStatusObserver::onReceive(const std::string& data) {
+    if (!strncmp(data.c_str(), "ping", sizeof("ping") - 1)) {
+        LOC_LOGd("ping received");
 
-    if ((socketFd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
-        LOC_LOGe("create socket error. reason:%s", strerror(errno));
-
-     } else {
-        const char* socketPath = XTRA_HAL_SOCKET_NAME ;
-        struct sockaddr_un addr = { .sun_family = AF_UNIX };
-        snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", socketPath);
-
-        if (::connect(socketFd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
-            LOC_LOGe("cannot connect to XTRA. reason:%s", strerror(errno));
-            if (::close(socketFd)) {
-                LOC_LOGe("close socket error. reason:%s", strerror(errno));
-            }
-            socketFd = -1;
-        }
-    }
-
-    return socketFd;
-}
-
-void XtraSystemStatusObserver::closeSocket(const int socketFd) {
-    if (socketFd >= 0) {
-        if(::close(socketFd)) {
-            LOC_LOGe("close socket error. reason:%s", strerror(errno));
-        }
+    } else {
+        LOC_LOGw("unknown event: %s", data.c_str());
     }
 }
 
diff --git a/gnss/XtraSystemStatusObserver.h b/gnss/XtraSystemStatusObserver.h
index fb16dc9..f8d3edd 100644
--- a/gnss/XtraSystemStatusObserver.h
+++ b/gnss/XtraSystemStatusObserver.h
@@ -34,11 +34,13 @@
 #ifdef USE_GLIB
 #include <LocThread.h>
 #endif
+#include <LocIpc.h>
 
 using namespace std;
 using loc_core::IOsObserver;
 using loc_core::IDataItemObserver;
 using loc_core::IDataItemCore;
+using loc_util::LocIpc;
 
 #ifdef USE_GLIB
 // XtraHalListenerSocket class
@@ -90,7 +92,7 @@ private:
 };
 #endif
 
-class XtraSystemStatusObserver : public IDataItemObserver {
+class XtraSystemStatusObserver : public IDataItemObserver, public LocIpc{
 public :
     // constructor & destructor
     inline XtraSystemStatusObserver(IOsObserver* sysStatObs, const MsgTask* msgTask):
@@ -99,9 +101,15 @@ public :
 #endif
             mSystemStatusObsrvr(sysStatObs), mMsgTask(msgTask) {
         subscribe(true);
+        startListeningNonBlocking(LOC_IPC_HAL);
+    }
+    inline XtraSystemStatusObserver() {
+        startListeningNonBlocking(LOC_IPC_HAL);
+    };
+    inline virtual ~XtraSystemStatusObserver() {
+        subscribe(false);
+        stopListening();
     }
-    inline XtraSystemStatusObserver() {};
-    inline virtual ~XtraSystemStatusObserver() { subscribe(false); }
 
     // IDataItemObserver overrides
     inline virtual void getName(string& name);
@@ -120,10 +128,9 @@ public :
     inline const MsgTask* getMsgTask() { return mMsgTask; }
     void subscribe(bool yes);
 
+    void onReceive(const std::string& data) override;
+
 private:
-    int createSocket();
-    void closeSocket(const int32_t socketFd);
-    bool sendEvent(const stringstream& event);
     IOsObserver*    mSystemStatusObsrvr;
     const MsgTask* mMsgTask;
 #ifdef USE_GLIB
diff --git a/utils/Android.mk b/utils/Android.mk
index 3f3b83a..1352076 100644
--- a/utils/Android.mk
+++ b/utils/Android.mk
@@ -26,7 +26,8 @@ LOCAL_SRC_FILES += \
     LocThread.cpp \
     MsgTask.cpp \
     loc_misc_utils.cpp \
-    loc_nmea.cpp
+    loc_nmea.cpp \
+    LocIpc.cpp
 
 # Flag -std=c++11 is not accepted by compiler when LOCAL_CLANG is set to true
 LOCAL_CFLAGS += \
diff --git a/utils/LocIpc.cpp b/utils/LocIpc.cpp
new file mode 100644
index 0000000..38889d0
--- /dev/null
+++ b/utils/LocIpc.cpp
@@ -0,0 +1,203 @@
+/* Copyright (c) 2017 The Linux Foundation. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ *       copyright notice, this list of conditions and the following
+ *       disclaimer in the documentation and/or other materials provided
+ *       with the distribution.
+ *     * Neither the name of The Linux Foundation, nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED "AS IS" AND ANY EXPRESS OR IMPLIED
+ * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT
+ * ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS
+ * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+ * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
+ * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
+ * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <sys/stat.h>
+#include <error.h>
+#include <errno.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <array>
+#include <platform_lib_log_util.h>
+#include <string>
+#include <vector>
+#include "gps_extended_c.h"
+#include "LocIpc.h"
+
+namespace loc_util {
+
+#ifdef LOG_TAG
+#undef LOG_TAG
+#endif
+#define LOG_TAG "LocSvc_LocIpc"
+
+#define LOC_MSG_BUF_LEN 1024
+#define LOC_MSG_HEAD "$MSGLEN$"
+
+class LocIpcRunnable : public LocRunnable {
+public:
+    LocIpcRunnable(LocIpc& locIpc, const std::string& ipcName)
+            : mLocIpc(locIpc), mIpcName(ipcName) {}
+    bool run() override {
+        if (!mLocIpc.startListeningBlocking(mIpcName)) {
+            LOC_LOGe("listen to socket failed");
+        }
+
+        return false;
+    }
+private:
+     LocIpc& mLocIpc;
+     const std::string mIpcName;
+};
+
+bool LocIpc::startListeningNonBlocking(const std::string& name) {
+    mRunnable.reset(new LocIpcRunnable(*this, name));
+    std::string threadName("LocIpc-");
+    threadName.append(name);
+    return mThread.start(threadName.c_str(), mRunnable.get());
+}
+
+bool LocIpc::startListeningBlocking(const std::string& name) {
+    int fd = socket(AF_UNIX, SOCK_DGRAM, 0);
+    if (fd < 0) {
+        LOC_LOGe("create socket error. reason:%s", strerror(errno));
+        return false;
+    }
+
+    if ((unlink(name.c_str()) < 0) && (errno != ENOENT)) {
+       LOC_LOGw("unlink socket error. reason:%s", strerror(errno));
+    }
+
+    struct sockaddr_un addr = { .sun_family = AF_UNIX };
+    snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", name.c_str());
+
+    umask(0157);
+
+    if (::bind(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+        LOC_LOGe("bind socket error. reason:%s", strerror(errno));
+        ::close(fd);
+        fd = -1;
+        return false;
+    }
+
+    mIpcFd = fd;
+
+    ssize_t nBytes = 0;
+    std::vector<char> buf(LOC_MSG_BUF_LEN);
+    while ((nBytes = ::recvfrom(mIpcFd, buf.data(), buf.size(), 0, NULL, NULL)) >= 0) {
+        if (nBytes == 0) {
+            continue;
+        }
+
+        std::string msg;
+        if (strncmp(buf.data(), LOC_MSG_HEAD, sizeof(LOC_MSG_HEAD) - 1)) {
+            // short message
+            msg.append(buf.data(), nBytes);
+            onReceive(msg);
+
+        } else {
+            // long message
+            size_t msgLen = 0;
+            sscanf(buf.data(), LOC_MSG_HEAD"%zu", &msgLen);
+            while (msg.length() < msgLen &&
+                    (nBytes = recvfrom(mIpcFd, buf.data(), buf.size(), 0, NULL, NULL)) >= 0) {
+                msg.append(buf.data(), nBytes);
+            }
+
+            if (nBytes >= 0) {
+                onReceive(msg);
+            } else {
+                break;
+            }
+        }
+    }
+
+    if (mStopRequested) {
+        mStopRequested = false;
+        return true;
+
+    } else {
+        LOC_LOGe("cannot read socket. reason:%s", strerror(errno));
+        (void)::close(mIpcFd);
+        mIpcFd = -1;
+        return false;
+    }
+}
+
+void LocIpc::stopListening() {
+    mStopRequested = true;
+
+    if (mIpcFd >= 0) {
+        if (::close(mIpcFd)) {
+            LOC_LOGe("cannot close socket:%s", strerror(errno));
+        }
+        mIpcFd = -1;
+    }
+}
+
+bool LocIpc::send(const char name[], const std::string& data) {
+    int fd  = ::socket(AF_UNIX, SOCK_DGRAM, 0);
+    if (fd < 0) {
+        LOC_LOGe("create socket error. reason:%s", strerror(errno));
+        return false;
+    }
+
+    struct sockaddr_un addr = { .sun_family = AF_UNIX };
+    snprintf(addr.sun_path, sizeof(addr.sun_path), "%s", name);
+
+    bool result = true;
+    if (data.length() <= LOC_MSG_BUF_LEN) {
+        if (::sendto(fd, data.c_str(), data.length(), 0,
+                (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+            LOC_LOGe("cannot send to socket. reason:%s", strerror(errno));
+            result = false;
+        }
+    } else {
+        std::string head = LOC_MSG_HEAD;
+        head.append(std::to_string(data.length()));
+        if (::sendto(fd, head.c_str(), head.length(), 0,
+                (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+            LOC_LOGe("cannot send to socket. reason:%s", strerror(errno));
+            result = false;
+        } else {
+            size_t sentBytes = 0;
+            while(sentBytes < data.length()) {
+                size_t partLen = data.length() - sentBytes;
+                if (partLen > LOC_MSG_BUF_LEN) {
+                    partLen = LOC_MSG_BUF_LEN;
+                }
+                ssize_t rv = ::sendto(fd, data.c_str() + sentBytes, partLen, 0,
+                        (struct sockaddr*)&addr, sizeof(addr));
+                if (rv < 0) {
+                    LOC_LOGe("cannot send to socket. reason:%s", strerror(errno));
+                    result = false;
+                    break;
+                }
+                sentBytes += rv;
+            }
+        }
+    }
+
+    (void)::close(fd);
+    return result;
+}
+
+}
diff --git a/utils/LocIpc.h b/utils/LocIpc.h
new file mode 100644
index 0000000..258fd42
--- /dev/null
+++ b/utils/LocIpc.h
@@ -0,0 +1,88 @@
+/* Copyright (c) 2017 The Linux Foundation. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ *       copyright notice, this list of conditions and the following
+ *       disclaimer in the documentation and/or other materials provided
+ *       with the distribution.
+ *     * Neither the name of The Linux Foundation, nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED "AS IS" AND ANY EXPRESS OR IMPLIED
+ * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT
+ * ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS
+ * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
+ * BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+ * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
+ * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
+ * IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#ifndef __LOC_SOCKET__
+#define __LOC_SOCKET__
+
+#include <string>
+#include <memory>
+#include <LocThread.h>
+
+namespace loc_util {
+
+class LocIpc {
+public:
+    inline LocIpc() : mIpcFd(-1), mStopRequested(false), mRunnable(nullptr) {}
+    inline virtual ~LocIpc() { stopListening(); }
+
+    // Listen for new messages in current thread. Calling this funciton will
+    // block current thread. The listening can be stopped by calling stopListening().
+    //
+    // Argument name is the path of the unix local socket to be listened.
+    // The function will return true on success, and false on failure.
+    bool startListeningBlocking(const std::string& name);
+
+    // Create a new LocThread and listen for new messages in it.
+    // Calling this function will return immediately and won't block current thread.
+    // The listening can be stopped by calling stopListening().
+    //
+    // Argument name is the path of the unix local socket to be be listened.
+    // The function will return true on success, and false on failure.
+    bool startListeningNonBlocking(const std::string& name);
+
+    // Stop listening to new messages.
+    void stopListening();
+
+    // Callback function for receiving incoming messages.
+    // Override this function in your derived class to process incoming messages.
+    // For each received message, this callback function will be called once.
+    // This callback function will be called in the calling thread of startListeningBlocking
+    // or in the new LocThread created by startListeningNonBlocking.
+    //
+    // Argument data contains the received message. You need to parse it.
+    virtual void onReceive(const std::string& /*data*/) {}
+
+    // Send out a message.
+    // Call this function to send a message in argument data to socket in argument name.
+    //
+    // Argument name contains the name of the target unix socket. data contains the
+    // message to be sent out. Convert your message to a string before calling this function.
+    // The function will return true on success, and false on failure.
+    bool send(const char name[], const std::string& data);
+
+private:
+    int mIpcFd;
+    bool mStopRequested;
+    LocThread mThread;
+    std::unique_ptr<LocRunnable> mRunnable;
+};
+
+}
+
+#endif //__LOC_SOCKET__
diff --git a/utils/gps_extended_c.h b/utils/gps_extended_c.h
index 7929a5a..a24e784 100644
--- a/utils/gps_extended_c.h
+++ b/utils/gps_extended_c.h
@@ -1309,6 +1309,9 @@ typedef void (*LocAgpsOpenResultCb)(bool isSuccess, AGpsExtType agpsType, const
 
 typedef void (*LocAgpsCloseResultCb)(bool isSuccess, AGpsExtType agpsType, void* userDataPtr);
 
+/* Shared resources of LocIpc */
+#define LOC_IPC_HAL "/data/vendor/location/socket_hal"
+#define LOC_IPC_XTRA "/data/vendor/location/xtra/socket_xtra"
 
 #ifdef __cplusplus
 }