ceph/src/jaegertracing/jaeger-client-cpp/src/jaegertracing/utils/HTTPTransporter.h

   1 /*
   2  * Copyright (c) 2019, The Jaeger Authors
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #ifndef JAEGERTRACING_UTILS_UDPCLIENT5_H
  18 #define JAEGERTRACING_UTILS_UDPCLIENT5_H
  19
  20 #include <iostream>
  21 #include <sstream>
  22 #include <stdexcept>
  23 #include <system_error>
  24
  25 #include "jaegertracing/Compilers.h"
  26
  27 #include "jaegertracing/utils/Transport.h"
  28
  29 #include "jaegertracing/net/IPAddress.h"
  30 #include "jaegertracing/net/Socket.h"
  31 #include "jaegertracing/net/http/Response.h"
  32
  33 #include <thrift/protocol/TBinaryProtocol.h>
  34 #include <thrift/transport/TBufferTransports.h>
  35 #include <thrift/transport/THttpClient.h>
  36
  37 namespace jaegertracing {
  38 namespace utils {
  39
  40 class HTTPTransporter : public Transport {
  41   public:
  42     HTTPTransporter(const net::URI& endpoint, int maxPacketSize);
  43
  44     ~HTTPTransporter() { close(); }
  45
  46     void emitBatch(const thrift::Batch& batch) override
  47     {
  48         // Resets the buffer to write a new batch
  49         _buffer->resetBuffer();
  50
  51         // Does the serialisation to Thrift
  52         auto oprot = _protocol.get();
  53         batch.write(oprot);
  54         oprot->writeMessageEnd();
  55         oprot->getTransport()->writeEnd();
  56         oprot->getTransport()->flush();
  57
  58         uint8_t* data = nullptr;
  59         uint32_t size = 0;
  60         _buffer->getBuffer(&data, &size);
  61
  62         // Sends the HTTP message
  63         const auto numWritten = ::send(_socket.handle(), reinterpret_cast<char*>(data), sizeof(uint8_t) * size, 0);
  64
  65         if (static_cast<unsigned>(numWritten) != size) {
  66             std::ostringstream oss;
  67             oss << "Failed to write message, numWritten=" << numWritten << ", size=" << size;
  68             throw std::system_error(errno, std::system_category(), oss.str());
  69         }
  70
  71         // Waits for response. Check that the server acknowledged
  72         // and returned a green status [200, 201, 202, 203, 204]
  73         net::http::Response response = net::http::read(_socket);
  74         if (response.statusCode() < 200 && response.statusCode() > 204) {
  75           std::ostringstream oss;
  76           oss << "Failed to write message, HTTP error " << response.statusCode() << response.reason();
  77           throw std::system_error(errno, std::system_category(), oss.str());
  78         }
  79     }
  80
  81     std::unique_ptr<apache::thrift::protocol::TProtocolFactory>
  82     protocolFactory() const override
  83     {
  84         return std::unique_ptr<apache::thrift::protocol::TProtocolFactory>(
  85             new apache::thrift::protocol::TBinaryProtocolFactory());
  86     }
  87
  88   private:
  89     std::shared_ptr<apache::thrift::transport::TMemoryBuffer> _buffer;
  90     net::IPAddress _serverAddr;
  91     std::shared_ptr<::apache::thrift::transport::THttpClient> _httpClient;
  92     std::shared_ptr<apache::thrift::protocol::TProtocol> _protocol;
  93
  94     static constexpr auto kHttpPacketMaxLength = 1024 * 1024; // 1MB
  95 };
  96
  97 }  // namespace utils
  98 }  // namespace jaegertracing
  99
 100 #endif  // JAEGERTRACING_UTILS_UDPCLIENT_H