AsyncPipe.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 <list>
#include <system_error>

#include <folly/io/IOBufQueue.h>
#include <folly/io/async/AsyncTransport.h>
#include <folly/io/async/DelayedDestruction.h>
#include <folly/io/async/EventHandler.h>

namespace folly {

class AsyncSocketException;

class AsyncPipeReader : public EventHandler,
                        public AsyncReader,
                        public DelayedDestruction {
 public:
  typedef std::
      unique_ptr<AsyncPipeReader, folly::DelayedDestruction::Destructor>
          UniquePtr;

  template <typename... Args>
  static UniquePtr newReader(Args&&... args) {
    return UniquePtr(new AsyncPipeReader(std::forward<Args>(args)...));
  }

  AsyncPipeReader(folly::EventBase* eventBase, int pipeFd)
      : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}

  void setReadCB(AsyncReader::ReadCallback* callback) override {
    if (callback == readCallback_) {
      return;
    }
    readCallback_ = callback;
    if (readCallback_ && !isHandlerRegistered()) {
      registerHandler(EventHandler::READ | EventHandler::PERSIST);
    } else if (!readCallback_ && isHandlerRegistered()) {
      unregisterHandler();
    }
  }

  AsyncReader::ReadCallback* getReadCallback() const override {
    return readCallback_;
  }

  void setCloseCallback(std::function<void(int)> closeCb) {
    closeCb_ = closeCb;
  }

 private:
  ~AsyncPipeReader() override;

  void handlerReady(uint16_t events) noexcept override;
  void failRead(const AsyncSocketException& ex);
  void close();

  int fd_;
  AsyncReader::ReadCallback* readCallback_{nullptr};
  std::function<void(int)> closeCb_;
};

class AsyncPipeWriter : public EventHandler,
                        public AsyncWriter,
                        public DelayedDestruction {
 public:
  typedef std::
      unique_ptr<AsyncPipeWriter, folly::DelayedDestruction::Destructor>
          UniquePtr;

  template <typename... Args>
  static UniquePtr newWriter(Args&&... args) {
    return UniquePtr(new AsyncPipeWriter(std::forward<Args>(args)...));
  }

  AsyncPipeWriter(folly::EventBase* eventBase, int pipeFd)
      : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}

  void write(
      std::unique_ptr<folly::IOBuf> iob,
      AsyncWriter::WriteCallback* wcb = nullptr);

  void setCloseCallback(std::function<void(int)> closeCb) {
    closeCb_ = closeCb;
  }

  bool closed() const {
    return (fd_ < 0 || closeOnEmpty_);
  }

  void closeOnEmpty();

  void closeNow();

  bool hasPendingWrites() const {
    return !queue_.empty();
  }

  // AsyncWriter methods
  void write(
      folly::AsyncWriter::WriteCallback* callback,
      const void* buf,
      size_t bytes,
      WriteFlags flags = WriteFlags::NONE) override {
    writeChain(callback, IOBuf::wrapBuffer(buf, bytes), flags);
  }
  void writev(
      folly::AsyncWriter::WriteCallback*,
      const iovec*,
      size_t,
      WriteFlags = WriteFlags::NONE) override {
    throw std::runtime_error("writev is not supported. Please use writeChain.");
  }
  void writeChain(
      folly::AsyncWriter::WriteCallback* callback,
      std::unique_ptr<folly::IOBuf>&& buf,
      WriteFlags flags = WriteFlags::NONE) override;

 private:
  void handlerReady(uint16_t events) noexcept override;
  void handleWrite();
  void failAllWrites(const AsyncSocketException& ex);

  int fd_;
  std::list<std::pair<folly::IOBufQueue, AsyncWriter::WriteCallback*>> queue_;
  bool closeOnEmpty_{false};
  std::function<void(int)> closeCb_;

  ~AsyncPipeWriter() override {
    closeNow();
  }
};

} // namespace folly