AsyncTransport.h

Go to the documentation of this file.

/*
 * Copyright 2014-present Facebook, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include <memory>

#include <folly/io/IOBuf.h>
#include <folly/io/async/AsyncSocketBase.h>
#include <folly/io/async/AsyncTransportCertificate.h>
#include <folly/io/async/DelayedDestruction.h>
#include <folly/io/async/EventBase.h>
#include <folly/portability/OpenSSL.h>
#include <folly/portability/SysUio.h>
#include <folly/ssl/OpenSSLPtrTypes.h>

constexpr bool kOpenSslModeMoveBufferOwnership =
#ifdef SSL_MODE_MOVE_BUFFER_OWNERSHIP
    true
#else
    false
#endif
    ;

namespace folly {

class AsyncSocketException;
class EventBase;
class SocketAddress;

/*
 * flags given by the application for write* calls
 */
enum class WriteFlags : uint32_t {
  NONE = 0x00,
  /*
   * Whether to delay the output until a subsequent non-corked write.
   * (Note: may not be supported in all subclasses or on all platforms.)
   */
  CORK = 0x01,
  /*
   * for a socket that has ACK latency enabled, it will cause the kernel
   * to fire a TCP ESTATS event when the last byte of the given write call
   * will be acknowledged.
   */
  EOR = 0x02,
  /*
   * this indicates that only the write side of socket should be shutdown
   */
  WRITE_SHUTDOWN = 0x04,
  /*
   * use msg zerocopy if allowed
   */
  WRITE_MSG_ZEROCOPY = 0x08,
};

/*
 * union operator
 */
inline WriteFlags operator|(WriteFlags a, WriteFlags b) {
  return static_cast<WriteFlags>(
      static_cast<uint32_t>(a) | static_cast<uint32_t>(b));
}

/*
 * compound assignment union operator
 */
inline WriteFlags& operator|=(WriteFlags& a, WriteFlags b) {
  a = a | b;
  return a;
}

/*
 * intersection operator
 */
inline WriteFlags operator&(WriteFlags a, WriteFlags b) {
  return static_cast<WriteFlags>(
      static_cast<uint32_t>(a) & static_cast<uint32_t>(b));
}

/*
 * compound assignment intersection operator
 */
inline WriteFlags& operator&=(WriteFlags& a, WriteFlags b) {
  a = a & b;
  return a;
}

/*
 * exclusion parameter
 */
inline WriteFlags operator~(WriteFlags a) {
  return static_cast<WriteFlags>(~static_cast<uint32_t>(a));
}

/*
 * unset operator
 */
inline WriteFlags unSet(WriteFlags a, WriteFlags b) {
  return a & ~b;
}

/*
 * inclusion operator
 */
inline bool isSet(WriteFlags a, WriteFlags b) {
  return (a & b) == b;
}

class AsyncTransport : public DelayedDestruction, public AsyncSocketBase {
 public:
  typedef std::unique_ptr<AsyncTransport, Destructor> UniquePtr;

  virtual void close() = 0;

  virtual void closeNow() = 0;

  virtual void closeWithReset() {
    closeNow();
  }

  virtual void shutdownWrite() = 0;

  virtual void shutdownWriteNow() = 0;

  virtual bool good() const = 0;

  virtual bool readable() const = 0;

  virtual bool writable() const {
    // By default return good() - leave it to implementers to override.
    return good();
  }

  virtual bool isPending() const {
    return readable();
  }

  virtual bool connecting() const = 0;

  virtual bool error() const = 0;

  virtual void attachEventBase(EventBase* eventBase) = 0;

  virtual void detachEventBase() = 0;

  virtual bool isDetachable() const = 0;

  virtual void setSendTimeout(uint32_t milliseconds) = 0;

  virtual uint32_t getSendTimeout() const = 0;

  virtual void getLocalAddress(SocketAddress* address) const = 0;

  SocketAddress getLocalAddress() const {
    SocketAddress addr;
    getLocalAddress(&addr);
    return addr;
  }

  void getAddress(SocketAddress* address) const override {
    getLocalAddress(address);
  }

  virtual void getPeerAddress(SocketAddress* address) const = 0;

  SocketAddress getPeerAddress() const {
    SocketAddress addr;
    getPeerAddress(&addr);
    return addr;
  }

  virtual ssl::X509UniquePtr getPeerCert() const {
    return nullptr;
  }

  virtual const X509* getSelfCert() const {
    return nullptr;
  }

  virtual const AsyncTransportCertificate* getPeerCertificate() const {
    return nullptr;
  }

  virtual const AsyncTransportCertificate* getSelfCertificate() const {
    return nullptr;
  }

  virtual std::string getApplicationProtocol() const noexcept {
    return "";
  }

  virtual std::string getSecurityProtocol() const {
    return "";
  }

  virtual bool isEorTrackingEnabled() const = 0;

  virtual void setEorTracking(bool track) = 0;

  virtual size_t getAppBytesWritten() const = 0;
  virtual size_t getRawBytesWritten() const = 0;
  virtual size_t getAppBytesReceived() const = 0;
  virtual size_t getRawBytesReceived() const = 0;

  class BufferCallback {
   public:
    virtual ~BufferCallback() {}
    virtual void onEgressBuffered() = 0;
    virtual void onEgressBufferCleared() = 0;
  };

  class ReplaySafetyCallback {
   public:
    virtual ~ReplaySafetyCallback() = default;

    virtual void onReplaySafe() = 0;
  };

  virtual bool isReplaySafe() const {
    return true;
  }

  virtual void setReplaySafetyCallback(ReplaySafetyCallback* callback) {
    if (callback) {
      CHECK(false) << "setReplaySafetyCallback() not supported";
    }
  }

 protected:
  ~AsyncTransport() override = default;
};

class AsyncReader {
 public:
  class ReadCallback {
   public:
    virtual ~ReadCallback() = default;

    virtual void getReadBuffer(void** bufReturn, size_t* lenReturn) = 0;

    virtual void readDataAvailable(size_t len) noexcept = 0;

    virtual bool isBufferMovable() noexcept {
      return false;
    }

    virtual size_t maxBufferSize() const {
      return 64 * 1024; // 64K
    }

    virtual void readBufferAvailable(
        std::unique_ptr<IOBuf> /*readBuf*/) noexcept {}

    virtual void readEOF() noexcept = 0;

    virtual void readErr(const AsyncSocketException& ex) noexcept = 0;
  };

  // Read methods that aren't part of AsyncTransport.
  virtual void setReadCB(ReadCallback* callback) = 0;
  virtual ReadCallback* getReadCallback() const = 0;

 protected:
  virtual ~AsyncReader() = default;
};

class AsyncWriter {
 public:
  class WriteCallback {
   public:
    virtual ~WriteCallback() = default;

    virtual void writeSuccess() noexcept = 0;

    virtual void writeErr(
        size_t bytesWritten,
        const AsyncSocketException& ex) noexcept = 0;
  };

  virtual void write(
      WriteCallback* callback,
      const void* buf,
      size_t bytes,
      WriteFlags flags = WriteFlags::NONE) = 0;

  virtual void writev(
      WriteCallback* callback,
      const iovec* vec,
      size_t count,
      WriteFlags flags = WriteFlags::NONE) = 0;

  virtual void writeChain(
      WriteCallback* callback,
      std::unique_ptr<IOBuf>&& buf,
      WriteFlags flags = WriteFlags::NONE) = 0;

 protected:
  virtual ~AsyncWriter() = default;
};

// Transitional intermediate interface. This is deprecated.
// Wrapper around folly::AsyncTransport, that includes read/write callbacks
class AsyncTransportWrapper : virtual public AsyncTransport,
                              virtual public AsyncReader,
                              virtual public AsyncWriter {
 public:
  using UniquePtr = std::unique_ptr<AsyncTransportWrapper, Destructor>;

  // Alias for inherited members from AsyncReader and AsyncWriter
  // to keep compatibility.
  using ReadCallback = AsyncReader::ReadCallback;
  using WriteCallback = AsyncWriter::WriteCallback;
  void setReadCB(ReadCallback* callback) override = 0;
  ReadCallback* getReadCallback() const override = 0;
  void write(
      WriteCallback* callback,
      const void* buf,
      size_t bytes,
      WriteFlags flags = WriteFlags::NONE) override = 0;
  void writev(
      WriteCallback* callback,
      const iovec* vec,
      size_t count,
      WriteFlags flags = WriteFlags::NONE) override = 0;
  void writeChain(
      WriteCallback* callback,
      std::unique_ptr<IOBuf>&& buf,
      WriteFlags flags = WriteFlags::NONE) override = 0;
  virtual const AsyncTransportWrapper* getWrappedTransport() const {
    return nullptr;
  }

  template <class T>
  const T* getUnderlyingTransport() const {
    const AsyncTransportWrapper* current = this;
    while (current) {
      auto sock = dynamic_cast<const T*>(current);
      if (sock) {
        return sock;
      }
      current = current->getWrappedTransport();
    }
    return nullptr;
  }

  template <class T>
  T* getUnderlyingTransport() {
    return const_cast<T*>(static_cast<const AsyncTransportWrapper*>(this)
                              ->getUnderlyingTransport<T>());
  }
};

} // namespace folly