src/IpcClient.cpp

   1 //
   2 // Open Service Platform
   3 // Copyright (c) 2012 Samsung Electronics Co., Ltd.
   4 //
   5 // Licensed under the Apache License, Version 2.0 (the License);
   6 // you may not use this file except in compliance with the License.
   7 // You may obtain a copy of the License at
   8 //
   9 //     http://www.apache.org/licenses/LICENSE-2.0
  10 //
  11 // Unless required by applicable law or agreed to in writing, software
  12 // distributed under the License is distributed on an "AS IS" BASIS,
  13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 // See the License for the specific language governing permissions and
  15 // limitations under the License.
  16 //
  17
  18 /**
  19  * @file        IpcClient.cpp
  20  * @brief       This is the implementation file for the IpcClient class.
  21  *
  22  */
  23
  24 #include <stdio.h>
  25 #include <errno.h>
  26 #include <sys/types.h>
  27 #include <sys/socket.h>
  28 #include <sys/un.h>
  29 #include <stdio.h>
  30 #include <stdlib.h>
  31 #include <unistd.h>
  32 #include <poll.h>
  33 #include <pthread.h>
  34 #include <fcntl.h>
  35
  36 #include <iostream>
  37 #include <queue>
  38 #include <map>
  39
  40 #include "message-port-log.h"
  41
  42 #include "IpcClient.h"
  43 #include "IIpcClientEventListener.h"
  44
  45 using namespace IPC;
  46 using namespace std;
  47
  48
  49 IpcClient::IpcClient(void)
  50         : __pReverseSource(NULL)
  51         , __pMutex(NULL)
  52         , __pListener(NULL)
  53 {
  54         __messageBuffer[0] = '\0';
  55 }
  56
  57 IpcClient::~IpcClient(void)
  58 {
  59         int fd = 0;
  60
  61         if (__pReverseSource != NULL)
  62         {
  63                 g_source_destroy(__pReverseSource);
  64                 g_source_unref(__pReverseSource);
  65                 __pReverseSource = NULL;
  66         }
  67
  68         while (__fds.size() > 0)
  69         {
  70                 fd = __fds.back();
  71                 __fds.pop_back();
  72
  73                 close(fd);
  74         }
  75
  76         pthread_mutex_destroy(__pMutex);
  77 }
  78
  79 int
  80 IpcClient::Construct(const string& serverName, const IIpcClientEventListener* pListener)
  81 {
  82         __name = serverName;
  83         __pListener = const_cast <IIpcClientEventListener*>(pListener);
  84
  85         pthread_mutex_t* pMutex = (pthread_mutex_t*) malloc(sizeof(pthread_mutex_t));
  86         if (pMutex == NULL)
  87         {
  88                 return -2;
  89         }
  90
  91         pthread_mutex_init(pMutex, NULL);
  92
  93         __pMutex = pMutex;
  94
  95         int ret = MakeConnection();
  96         if (ret != 0)
  97         {
  98                 return ret;
  99         }
 100
 101         if (__pListener)
 102         {
 103                 ret = MakeConnection(true);
 104                 if (ret != 0)
 105                 {
 106                         return ret;
 107                 }
 108         }
 109
 110         return 0;
 111
 112 }
 113
 114 string
 115 IpcClient::GetName(void) const
 116 {
 117         return __name;
 118 }
 119
 120 struct HelloMessage
 121 {
 122         int pid;
 123         bool reverse;
 124 };
 125
 126 int
 127 IpcClient::MakeConnection(bool forReverse)
 128 {
 129         int ret = 0;
 130
 131         size_t socketNameLength = 0;
 132         string socketName;
 133
 134         socketName.append("/tmp/");
 135         socketName.append(__name);
 136
 137         socketNameLength = socketName.size() + 1;
 138
 139         HelloMessage helloMessage = {0, false};
 140
 141         helloMessage.pid = getpid();
 142         helloMessage.reverse = forReverse;
 143
 144         struct sockaddr_un server;
 145
 146         bzero(&server, sizeof(server));
 147         server.sun_family = AF_UNIX;
 148         strncpy(server.sun_path, socketName.c_str(), socketNameLength);
 149         socklen_t serverLen = sizeof(server);
 150
 151         int client = socket(AF_UNIX, SOCK_STREAM, 0);
 152         if (client < 0)
 153         {
 154                 _LOGE("Failed to create a socket : %s.", strerror(errno));
 155                 return -1;
 156         }
 157
 158         int flags = fcntl(client, F_GETFL, 0);
 159         ret = fcntl(client, F_SETFL, flags | O_NONBLOCK);
 160         if (ret != 0)
 161         {
 162                 _LOGE("Failed to set file status flags (%d, %s).", errno, strerror(errno));
 163                 goto CATCH;
 164         }
 165
 166         ret = connect(client, (struct sockaddr*) &server, serverLen);
 167         if (ret != 0)
 168         {
 169                 if (errno != EINPROGRESS)
 170                 {
 171                         _LOGE("Failed to connect to server(%s) : %s", socketName.c_str(), strerror(errno));
 172                         goto CATCH;
 173                 }
 174
 175                 fd_set rset;
 176                 fd_set wset;
 177                 struct timeval timeout;
 178                 int length = 0;
 179                 int error = 0;
 180                 socklen_t socketLength = 0;
 181
 182                 FD_ZERO(&rset);
 183                 FD_SET(client, &rset);
 184                 wset = rset;
 185                 timeout.tv_sec = 10;
 186                 timeout.tv_usec = 0;
 187
 188                 while (true)
 189                 {
 190                         ret = select(client+1, &rset, &wset, NULL, &timeout);
 191                         if (ret < 0)
 192                         {
 193                                 _LOGE("Failed to connect due to system error : %s.", strerror(errno));
 194                                 if (errno != EINTR)
 195                                 {
 196                                         goto CATCH;
 197                                 }
 198
 199                                 continue;
 200                         }
 201                         else if (ret == 0)
 202                         {
 203                                 _LOGE("Failed to connect due to timeout.");
 204                                 goto CATCH;
 205                         }
 206                         else
 207                         {
 208                                 break;
 209                         }
 210                 }
 211
 212                 if (FD_ISSET(client, &rset) || FD_ISSET(client, &wset))
 213                 {
 214                         length = sizeof(error);
 215                         ret = getsockopt(client, SOL_SOCKET, SO_ERROR, &error, &socketLength);
 216                         if (ret < 0)
 217                         {
 218                                 _LOGE("Failed to connect to server(%s) : %s", socketName.c_str(), strerror(errno));
 219                                 goto CATCH;
 220                         }
 221                 }
 222                 else
 223                 {
 224                         _LOGE("Failed to connect due to system error.");
 225                         goto CATCH;
 226                 }
 227         }
 228
 229         ret = fcntl(client, F_SETFL, flags);
 230         if (ret < 0)
 231         {
 232                 _LOGE("Failed to set file status flags (%d, %s).", errno, strerror(errno));
 233                 goto CATCH;
 234         }
 235
 236         ret = write(client, &helloMessage, sizeof(helloMessage));
 237         if (ret < 0)
 238         {
 239                 goto CATCH;
 240         }
 241
 242         if (forReverse)
 243         {
 244                 GError* pGError = NULL;
 245                 GSource* pGSource = NULL;
 246
 247                 GIOChannel* pChannel = g_io_channel_unix_new(client);
 248                 GMainContext* pGMainContext = g_main_context_default();
 249
 250                 g_io_channel_set_encoding(pChannel, NULL, &pGError);
 251                 g_io_channel_set_flags(pChannel, G_IO_FLAG_NONBLOCK, &pGError);
 252                 g_io_channel_set_close_on_unref(pChannel, TRUE);
 253
 254                 pGSource = g_io_create_watch(pChannel, (GIOCondition) (G_IO_IN | G_IO_ERR | G_IO_NVAL | G_IO_HUP));
 255                 g_source_set_callback(pGSource, (GSourceFunc) OnReadMessage, this, NULL);
 256                 g_source_attach(pGSource, pGMainContext);
 257
 258                 g_io_channel_unref(pChannel);
 259                 __pReverseSource = pGSource;
 260         }
 261         else
 262         {
 263                 ReleaseFd(client);
 264         }
 265
 266         return 0;
 267
 268 CATCH:
 269         if (client != -1)
 270         {
 271                 close(client);
 272         }
 273
 274         return -1;
 275
 276 }
 277
 278 gboolean
 279 IpcClient::OnReadMessage(GIOChannel* source, GIOCondition condition, gpointer data)
 280 {
 281         IpcClient* pIpcClient = (IpcClient*) data;
 282
 283         return pIpcClient->HandleReceivedMessage(source, condition);
 284 }
 285
 286 gboolean
 287 IpcClient::HandleReceivedMessage(GIOChannel* source, GIOCondition condition)
 288 {
 289         GError* pGError = NULL;
 290         GIOStatus status;
 291         IPC::Message* pMessage = NULL;
 292
 293         if (condition & G_IO_HUP)
 294         {
 295                 _LOGD("G_IO_HUP, the connection is closed.");
 296
 297                 g_source_destroy(__pReverseSource);
 298                 g_source_unref(__pReverseSource);
 299                 __pReverseSource = NULL;
 300
 301                 if (__pListener)
 302                 {
 303                         __pListener->OnIpcServerDisconnected(*this);
 304                 }
 305
 306                 return FALSE;
 307         }
 308         else if (condition & G_IO_IN)
 309         {
 310                 gsize readSize = 0;
 311                 const char* pStart = NULL;
 312                 const char* pEnd = NULL;
 313                 const char* pEndOfMessage = NULL;
 314
 315                 while (true)
 316                 {
 317                         pGError = NULL;
 318                         status = g_io_channel_read_chars(source, (char*) __messageBuffer, __MAX_MESSAGE_BUFFER_SIZE, &readSize, &pGError);
 319                         if (status == G_IO_STATUS_EOF || status == G_IO_STATUS_ERROR)
 320                         {
 321                                 if (status == G_IO_STATUS_EOF)
 322                                 {
 323                                         _LOGD("G_IO_STATUS_EOF, the connection is closed.");
 324                                 }
 325                                 else
 326                                 {
 327                                         _LOGD("G_IO_STATUS_ERROR, the connection is closed.");
 328                                 }
 329
 330                                 pGError = NULL;
 331
 332                                 g_io_channel_shutdown(source, FALSE, &pGError);
 333
 334                                 g_source_destroy(__pReverseSource);
 335                                 g_source_unref(__pReverseSource);
 336                                 __pReverseSource = NULL;
 337
 338                                 if (__pListener)
 339                                 {
 340                                         __pListener->OnIpcServerDisconnected(*this);
 341                                 }
 342
 343                                 return FALSE;
 344                         }
 345
 346                         if (readSize == 0)
 347                         {
 348                                 break;
 349                         }
 350
 351                         if (__pending.empty())
 352                         {
 353                                 pStart = __messageBuffer;
 354                                 pEnd = pStart + readSize;
 355                         }
 356                         else
 357                         {
 358                                 __pending.append(__messageBuffer, readSize);
 359                                 pStart = __pending.data();
 360                                 pEnd = pStart + __pending.size();
 361                         }
 362
 363                         while (true)
 364                         {
 365                                 pEndOfMessage = IPC::Message::FindNext(pStart, pEnd);
 366                                 if (pEndOfMessage == NULL)
 367                                 {
 368                                         __pending.assign(pStart, pEnd - pStart);
 369                                         break;
 370                                 }
 371
 372                                 pMessage = new (std::nothrow) IPC::Message(pStart, pEndOfMessage - pStart);
 373                                 if (pMessage == NULL)
 374                                 {
 375                                         _LOGE("The memory is insufficient");
 376                                         return -2;
 377                                 }
 378
 379                                 if (__pListener)
 380                                 {
 381                                         __pListener->OnIpcResponseReceived(*this, *pMessage);
 382                                 }
 383
 384                                 delete pMessage;
 385
 386                                 pStart = pEndOfMessage;
 387                         }
 388                 }
 389         }
 390         else
 391         {
 392                 // empty statement.
 393         }
 394
 395         return TRUE;
 396 }
 397
 398 int
 399 IpcClient::AcquireFd(void)
 400 {
 401         int ret = 0;
 402         int fd = -1;
 403
 404         while (fd == -1)
 405         {
 406                 pthread_mutex_lock(__pMutex);
 407                 if (__fds.size() == 0)
 408                 {
 409                         pthread_mutex_unlock(__pMutex);
 410                         ret = MakeConnection(false);
 411                         if (ret < 0)
 412                         {
 413                                 _LOGE("Failed to connect to the server.");
 414                                 return -1;
 415                         }
 416
 417                         continue;
 418                 }
 419
 420                 fd = __fds.back();
 421                 __fds.pop_back();
 422
 423                 pthread_mutex_unlock(__pMutex);
 424         }
 425
 426         return fd;
 427 }
 428
 429 void
 430 IpcClient::ReleaseFd(int fd)
 431 {
 432         pthread_mutex_lock(__pMutex);
 433
 434         __fds.push_back(fd);
 435
 436         pthread_mutex_unlock(__pMutex);
 437 }
 438
 439 int
 440 IpcClient::SendAsync(IPC::Message* pMessage)
 441 {
 442         char* pData = (char*) pMessage->data();
 443         int remain = pMessage->size();
 444         int fd = AcquireFd();
 445         if (fd == -1)
 446         {
 447                 _LOGE("Failed to get fd.");
 448                 return -1;
 449         }
 450
 451         int written = 0;
 452         while (remain > 0)
 453         {
 454                 written = write(fd, (char*) pData, remain);
 455                 remain -= written;
 456                 pData += written;
 457         }
 458
 459         ReleaseFd(fd);
 460
 461         return 0;
 462 }
 463
 464 int
 465 IpcClient::SendSync(IPC::Message* pMessage)
 466 {
 467         IPC::Message* pReply = NULL;
 468         MessageReplyDeserializer* pReplyDeserializer = NULL;
 469         IPC::SyncMessage* pSyncMessage = dynamic_cast <IPC::SyncMessage*>(pMessage);
 470         if (pSyncMessage == NULL)
 471         {
 472                 _LOGE("pMessage is not a sync message.");
 473                 return -1;
 474         }
 475
 476         int messageId = SyncMessage::GetMessageId(*pSyncMessage);
 477
 478         int fd = AcquireFd();
 479         if (fd < 0)
 480         {
 481                 _LOGE("Failed to get fd.");
 482                 return -1;
 483         }
 484
 485         char* pData = (char*) pSyncMessage->data();
 486         int remain = pSyncMessage->size();
 487         int written = 0;
 488
 489         while (remain > 0)
 490         {
 491                 written = write(fd, (char*) pData, remain);
 492                 remain -= written;
 493                 pData += written;
 494         }
 495
 496         // Wait reply
 497         struct pollfd pfd;
 498
 499         pfd.fd = fd;
 500         pfd.events = POLLIN | POLLRDHUP;
 501         pfd.revents = 0;
 502
 503         char buffer[1024];
 504         std::string message;
 505         int readSize = 0;
 506         char* pEndOfMessage = NULL;
 507
 508         while (true)
 509         {
 510                 poll(&pfd, 1, -1);
 511
 512                 if (pfd.revents & POLLRDHUP)
 513                 {
 514                         return -1;
 515                 }
 516
 517                 if (pfd.revents & POLLIN)
 518                 {
 519                         readSize = read(fd, buffer, 1024);
 520                 }
 521
 522                 message.append(buffer, readSize);
 523
 524                 pEndOfMessage = (char*) IPC::Message::FindNext(message.data(), message.data() + message.size());
 525                 if (pEndOfMessage)
 526                 {
 527                         pReply = new (std::nothrow) IPC::Message(message.data(), pEndOfMessage - message.data());
 528                         if (pReply == NULL)
 529                         {
 530                                 _LOGE("The memory is insufficient.");
 531                                 return -2;
 532                         }
 533
 534                         break;
 535                 }
 536         }
 537
 538         pReplyDeserializer = pSyncMessage->GetReplyDeserializer();
 539         pReplyDeserializer->SerializeOutputParameters(*pReply);
 540
 541         delete pReply;
 542         delete pReplyDeserializer;
 543
 544         ReleaseFd(fd);
 545
 546         return 0;
 547 }
 548
 549 int
 550 IpcClient::Send(IPC::Message* pMessage)
 551 {
 552         int ret = 0;
 553
 554         if (pMessage->is_sync())
 555         {
 556                 ret = SendSync(pMessage);
 557         }
 558         else
 559         {
 560                 ret = SendAsync(pMessage);
 561         }
 562
 563         return ret;
 564 }
 565
 566 int
 567 IpcClient::SendRequest(IPC::Message* pMessage)
 568 {
 569         return Send(pMessage);
 570 }
 571
 572 int
 573 IpcClient::SendRequest(const IPC::Message& message)
 574 {
 575         int ret = 0;
 576
 577         if (message.is_sync())
 578         {
 579                 ret = SendSync(const_cast<IPC::Message*>(&message));
 580         }
 581         else
 582         {
 583                 ret = SendAsync(const_cast<IPC::Message*>(&message));
 584         }
 585
 586         return ret;
 587 }
 588