-
-bool STG_CLIENT::reconnect()
-{
- if (stgClient == NULL)
- {
- RadLog("Connection is not configured.");
- return false;
- }
- if (!stgClient->stop())
- {
- RadLog("Failed to drop previous connection.");
- return false;
- }
- try {
- STG_CLIENT* old = stgClient;
- stgClient = new STG_CLIENT(*old);
- delete old;
- return true;
- } catch (const ChannelConfig::Error& ex) {
- // TODO: Log it
- RadLog("Client configuration error: %s.", ex.what());
- }
- return false;
-}
-
-STG_CLIENT::Impl::Impl(const std::string& address, Callback callback, void* data)
- : m_config(address),
- m_sock(connect()),
- m_running(false),
- m_stopped(true),
- m_lastPing(time(NULL)),
- m_lastActivity(m_lastPing),
- m_callback(callback),
- m_data(data),
- m_parser(&STG_CLIENT::Impl::process, this),
- m_connected(true)
-{
- int res = pthread_create(&m_thread, NULL, &STG_CLIENT::Impl::run, this);
- if (res != 0)
- throw Error("Failed to create thread: " + std::string(strerror(errno)));
-}
-
-STG_CLIENT::Impl::Impl(const Impl& rhs)
- : m_config(rhs.m_config),
- m_sock(connect()),
- m_running(false),
- m_stopped(true),
- m_lastPing(time(NULL)),
- m_lastActivity(m_lastPing),
- m_callback(rhs.m_callback),
- m_data(rhs.m_data),
- m_parser(&STG_CLIENT::Impl::process, this),
- m_connected(true)
-{
- int res = pthread_create(&m_thread, NULL, &STG_CLIENT::Impl::run, this);
- if (res != 0)
- throw Error("Failed to create thread: " + std::string(strerror(errno)));
-}
-
-STG_CLIENT::Impl::~Impl()
-{
- stop();
- shutdown(m_sock, SHUT_RDWR);
- close(m_sock);
-}
-
-bool STG_CLIENT::Impl::stop()
-{
- m_connected = false;
-
- if (m_stopped)
- return true;
-
- m_running = false;
-
- for (size_t i = 0; i < 25 && !m_stopped; i++) {
- struct timespec ts = {0, 200000000};
- nanosleep(&ts, NULL);
- }
-
- if (m_stopped) {
- pthread_join(m_thread, NULL);
- return true;
- }
-
- return false;
-}
-
-bool STG_CLIENT::Impl::request(TYPE type, const std::string& userName, const std::string& password, const PAIRS& pairs)
-{
- MapGen map;
- for (PAIRS::const_iterator it = pairs.begin(); it != pairs.end(); ++it)
- map.add(it->first, new StringGen(it->second));
- map.add("Radius-Username", new StringGen(userName));
- map.add("Radius-Userpass", new StringGen(password));
-
- PacketGen gen("data");
- gen.add("stage", toStage(type))
- .add("pairs", map);
-
- m_lastPing = time(NULL);
-
- return generate(gen, &STG_CLIENT::Impl::write, this);
-}
-
-void STG_CLIENT::Impl::runImpl()
-{
- m_running = true;
-
- while (m_running) {
- fd_set fds;
-
- FD_ZERO(&fds);
- FD_SET(m_sock, &fds);
-
- struct timeval tv;
- tv.tv_sec = 0;
- tv.tv_usec = 500000;
-
- int res = select(m_sock + 1, &fds, NULL, NULL, &tv);
- if (res < 0)
- {
- if (errno == EINTR)
- continue;
- RadLog("'select' is failed: %s", strerror(errno));
- //m_error = std::string("'select' is failed: '") + strerror(errno) + "'.";
- //m_logger(m_error);
- break;
- }
-
- if (!m_running)
- break;
-
- if (res > 0)
- {
- if (FD_ISSET(m_sock, &fds))
- m_running = read();
- }
- else
- m_running = tick();
- }
-
- m_connected = false;
- m_stopped = true;
-}
-
-int STG_CLIENT::Impl::connect()
-{
- 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)
- {
- impl.m_connected = false;
- RadLog("Failed to write data: %s.", strerror(errno));
- //conn.m_logger("Failed to write pong to '" + conn.m_remote + "': " + strerror(errno));
- return false;
- }
- size -= res;
- }
- return true;
-}
-
-void* STG_CLIENT::Impl::run(void* data)
-{
- Impl& impl = *static_cast<Impl*>(data);
- impl.runImpl();
- return NULL;
-}
git.stg.codes / stg.git / blobdiff