+ if (m_config.transport == "tcp")
+ return connectTCP();
+ else if (m_config.transport == "unix")
+ return connectUNIX();
+ throw Error("Invalid transport type: '" + m_config.transport + "'. Should be 'tcp' or 'unix'.");
+}
+
+int STG_CLIENT::Impl::connectTCP()
+{
+ addrinfo hints;
+ memset(&hints, 0, sizeof(addrinfo));
+
+ hints.ai_family = AF_INET; /* Allow IPv4 */
+ hints.ai_socktype = SOCK_STREAM; /* Stream socket */
+ hints.ai_flags = 0; /* For wildcard IP address */
+ hints.ai_protocol = 0; /* Any protocol */
+ hints.ai_canonname = NULL;
+ hints.ai_addr = NULL;
+ hints.ai_next = NULL;
+
+ addrinfo* ais = NULL;
+ int res = getaddrinfo(m_config.address.c_str(), m_config.portStr.c_str(), &hints, &ais);
+ if (res != 0)
+ throw Error("Error resolvin address '" + m_config.address + "': " + gai_strerror(res));
+
+ for (addrinfo* ai = ais; ai != NULL; ai = ai->ai_next)
+ {
+ int fd = socket(AF_INET, SOCK_STREAM, 0);
+ if (fd == -1)
+ {
+ Error error(std::string("Error creating TCP socket: ") + strerror(errno));
+ freeaddrinfo(ais);
+ throw error;
+ }
+ if (::connect(fd, ai->ai_addr, ai->ai_addrlen) == -1)
+ {
+ shutdown(fd, SHUT_RDWR);
+ close(fd);
+ RadLog("'connect' is failed: %s", strerror(errno));
+ // TODO: log it.
+ continue;
+ }
+ freeaddrinfo(ais);
+ return fd;
+ }
+
+ freeaddrinfo(ais);
+
+ throw Error("Failed to resolve '" + m_config.address);
+};
+
+int STG_CLIENT::Impl::connectUNIX()
+{
+ int fd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (fd == -1)
+ throw Error(std::string("Error creating UNIX socket: ") + strerror(errno));
+ struct sockaddr_un addr;
+ memset(&addr, 0, sizeof(addr));
+ addr.sun_family = AF_UNIX;
+ strncpy(addr.sun_path, m_config.address.c_str(), m_config.address.length());
+ if (::connect(fd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) == -1)
+ {
+ Error error(std::string("Error connecting UNIX socket: ") + strerror(errno));
+ shutdown(fd, SHUT_RDWR);
+ close(fd);
+ throw error;
+ }
+ return fd;
+}
+
+bool STG_CLIENT::Impl::read()
+{
+ static std::vector<char> buffer(1024);
+ ssize_t res = ::read(m_sock, buffer.data(), buffer.size());
+ if (res < 0)
+ {
+ RadLog("Failed to read data: ", strerror(errno));
+ //m_logger("Failed to read data from '" + m_remote + "': " + strerror(errno));
+ return false;
+ }
+ m_lastActivity = time(NULL);
+ RadLog("Read %d bytes.\n%s\n", res, std::string(buffer.data(), res).c_str());
+ if (res == 0)
+ {
+ m_parser.last();
+ return false;
+ }
+ return m_parser.append(buffer.data(), res);
+}
+
+bool STG_CLIENT::Impl::tick()
+{
+ time_t now = time(NULL);
+ if (difftime(now, m_lastActivity) > CONN_TIMEOUT)
+ {
+ int delta = difftime(now, m_lastActivity);
+ RadLog("Connection timeout: %d sec.", delta);
+ //m_logger("Connection to " + m_remote + " timed out.");
+ return false;
+ }
+ if (difftime(now, m_lastPing) > PING_TIMEOUT)
+ {
+ int delta = difftime(now, m_lastPing);
+ RadLog("Ping timeout: %d sec. Sending ping...", delta);
+ sendPing();
+ }
+ return true;
+}
+
+void STG_CLIENT::Impl::process(void* data)
+{
+ Impl& impl = *static_cast<Impl*>(data);
+ switch (impl.m_parser.packet())
+ {
+ case PING:
+ impl.processPing();
+ return;
+ case PONG:
+ impl.processPong();
+ return;
+ case DATA:
+ impl.processData();
+ return;
+ }
+ RadLog("Received invalid packet type: '%s'.", impl.m_parser.packetStr().c_str());
+}
+
+void STG_CLIENT::Impl::processPing()
+{
+ RadLog("Got ping, sending pong.");
+ sendPong();
+}
+
+void STG_CLIENT::Impl::processPong()
+{
+ RadLog("Got pong.");
+ m_lastActivity = time(NULL);
+}
+
+void STG_CLIENT::Impl::processData()
+{
+ RESULT data;
+ RadLog("Got data.");
+ for (PairsParser::Pairs::const_iterator it = m_parser.reply().begin(); it != m_parser.reply().end(); ++it)
+ data.reply.push_back(std::make_pair(it->first, it->second));
+ for (PairsParser::Pairs::const_iterator it = m_parser.modify().begin(); it != m_parser.modify().end(); ++it)
+ data.modify.push_back(std::make_pair(it->first, it->second));
+ m_callback(m_data, data, m_parser.result());
+}
+
+bool STG_CLIENT::Impl::sendPing()
+{
+ PacketGen gen("ping");
+
+ m_lastPing = time(NULL);
+
+ return generate(gen, &STG_CLIENT::Impl::write, this);
+}
+
+bool STG_CLIENT::Impl::sendPong()
+{
+ PacketGen gen("pong");
+
+ m_lastPing = time(NULL);
+
+ return generate(gen, &STG_CLIENT::Impl::write, this);
+}
+
+bool STG_CLIENT::Impl::write(void* data, const char* buf, size_t size)
+{
+ RadLog("Sending JSON:");
+ std::string json(buf, size);
+ RadLog("%s", json.c_str());
+ STG_CLIENT::Impl& impl = *static_cast<STG_CLIENT::Impl*>(data);
+ while (size > 0)
+ {
+ ssize_t res = ::send(impl.m_sock, buf, size, MSG_NOSIGNAL);
+ if (res < 0)
+ {
+ RadLog("Failed to write data: %s.", strerror(errno));
+ //conn.m_logger("Failed to write pong to '" + conn.m_remote + "': " + strerror(errno));
+ return false;
git.stg.codes / stg.git / blobdiff