Method send

src/posix/libevent/libevent_ssl_socket.cpp:755–814

Future<size_t> LibeventSSLSocketImpl::send(const char* data, size_t size)
{
  // Optimistically construct a 'SendRequest' and future.
  Owned<SendRequest> request(new SendRequest(size));
  Future<size_t> future = request->promise.future();

  // We don't add an 'onDiscard' continuation to send because we can
  // not accurately detect how many bytes have been sent. Once we pass
  // the data to the bufferevent, there is the possibility that parts
  // of it have been sent. Another reason is that if we send partial
  // messages (discard only a part of the data), then it is likely
  // that the receiving end will fail parsing the message.

  // Assign 'send_request' under lock, fail on error.
  synchronized (lock) {
    if (send_request.get() != nullptr) {
      return Failure("Socket is already sending");
    }
    std::swap(request, send_request);
  }

  evbuffer* buffer = CHECK_NOTNULL(evbuffer_new());

  int result = evbuffer_add(buffer, data, size);
  CHECK_EQ(0, result);

  // Extend the life-time of 'this' through the execution of the
  // lambda in the event loop. Note: The 'self' needs to be explicitly
  // captured because we're not using it in the body of the lambda. We
  // can use a 'shared_ptr' because run_in_event_loop is guaranteed to
  // execute.
  auto self = shared(this);

  run_in_event_loop(
      [self, buffer]() {
        CHECK(__in_event_loop__);
        CHECK(self);

        // Check if the socket is closed or the write end has
        // encountered an error in the interim (i.e. we received
        // a BEV_EVENT_ERROR with BEV_EVENT_WRITING).
        bool write = false;

        synchronized (self->lock) {
          if (self->send_request.get() != nullptr) {
            write = true;
          }
        }

        if (write) {
          int result = bufferevent_write_buffer(self->bev, buffer);
          CHECK_EQ(0, result);
        }

        evbuffer_free(buffer);
      },
      DISALLOW_SHORT_CIRCUIT);