#include "SocketConnection.h" #include "lib/jetfix/store/session_name.h" #include "lib/logging/logging.h" #include "lib/tracing/bts_trace.h" #include "lib/utils/TimeStamp.h" #include "lib/utils/bts_assert.h" #include "lib/utils/helper_macros.h" #include "lib/tracing/PerfLog.h" #include #include #include #include #include #include #include namespace bts { namespace jetfix { SocketConnection::SocketConnection(bts::config::JetFIX::Connection const &cfg, trace::PerfLog &perf_log) : _fd(-1), _cfg(cfg), _log(cfg.log_fix_stream()), _in_log(), _out_log(), _perf_log(perf_log) { if (_log) { _in_log.open(std::string("/tmp/fix_raw_read_log_") + session_name(cfg), std::ios_base::app); _out_log.open(std::string("/tmp/fix_raw_write_log_") + session_name(cfg), std::ios_base::app); } } SocketConnection::~SocketConnection() { if (_fd > 0) close(_fd); } bool SocketConnection::connect() { bts_assert(_fd == -1); _fd = socket(AF_INET, SOCK_STREAM, 0); if (_fd < 0) { auto er = errno; char msg[256]; LOG(Jetfix, Error, "Failed to allocate socket when connecting to {}:{} errno:{} msg: {}", _cfg.socket_connect_host_ip(), _cfg.socket_connect_port(), er, std::string(strerror_r(er, msg, sizeof(msg)))); _fd = -1; return false; } struct sockaddr_in serv_addr; memset(&serv_addr, '0', sizeof(serv_addr)); serv_addr.sin_family = AF_INET; serv_addr.sin_port = htons(_cfg.socket_connect_port()); auto result = inet_pton(AF_INET, _cfg.socket_connect_host_ip().c_str(), &serv_addr.sin_addr); bts_assert(result > 0); result = ::connect(_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)); if (result < 0) { LOG(Jetfix, Info, "Failed to connect to {}:{}", _cfg.socket_connect_host_ip(), _cfg.socket_connect_port()); return false; } else { LOG(Jetfix, Info, "Connected to {}:{}", _cfg.socket_connect_host_ip(), _cfg.socket_connect_port()); int flag = 1; result = ::setsockopt(_fd, IPPROTO_TCP, TCP_NODELAY, (char *)&flag, sizeof(int)); bts_assert(result == 0); return true; } } void SocketConnection::shutdown() { bts_assert(_fd >= 0); ::shutdown(_fd, SHUT_RDWR); } int SocketConnection::read(char *to_buffer, size_t read_count) { bts_assert(_fd >= 0); auto result = ::read(_fd, to_buffer, read_count); if (_log && result > 0) { _in_log.write(to_buffer, result); _in_log.flush(); } if (result < 0) { auto er = errno; char msg[256]; LOG(Jetfix, Warn, "Could not read from jetfix connection. {}", std::string(strerror_r(er, msg, sizeof(msg)))); } return result; } int SocketConnection::write(WriteVec const &vec, size_t vec_size, trace::CorID trace) { bts_assert(_fd >= 0); struct iovec iv[MaxWriteVecSz]; for (size_t i = 0; i < vec_size; i++) { iv[i].iov_base = (void *)vec[i].first; iv[i].iov_len = vec[i].second; } _perf_log.log(trace,"before_fix_socket_w"); auto result = ::writev(_fd, iv, (int)vec_size); _perf_log.log(trace,"after_fix_socket_w"); if (trace.sn) _perf_log.flush(); if (_log && result > 0) { for (size_t i = 0; i < vec_size; i++) _out_log.write(vec[i].first, vec[i].second); _out_log.flush(); } if (result < 0) { auto er = errno; char msg[256]; LOG(Jetfix, Warn, "Could not write to jetfix connection. ", std::string(strerror_r(er, msg, sizeof(msg)))); } return result; } } // namespace jetfix } // namespace bts