wire_reader_impl.h

Go to the documentation of this file.

// Copyright 2021 gRPC authors.
//
// 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.
 
#ifndef GRPC_CORE_EXT_TRANSPORT_BINDER_WIRE_FORMAT_WIRE_READER_IMPL_H
#define GRPC_CORE_EXT_TRANSPORT_BINDER_WIRE_FORMAT_WIRE_READER_IMPL_H
 
#include <grpc/support/port_platform.h>
 
#include <memory>
#include <utility>
 
#include "absl/container/flat_hash_map.h"
#include "absl/synchronization/notification.h"
 
#include <grpcpp/security/binder_security_policy.h>
 
#include "src/core/ext/transport/binder/utils/transport_stream_receiver.h"
#include "src/core/ext/transport/binder/wire_format/binder.h"
#include "src/core/ext/transport/binder/wire_format/wire_reader.h"
#include "src/core/ext/transport/binder/wire_format/wire_writer.h"
 
namespace grpc_binder {
 
class WireReaderImpl : public WireReader {
 public:
  WireReaderImpl(
      std::shared_ptr<TransportStreamReceiver> transport_stream_receiver,
      bool is_client,
      std::shared_ptr<grpc::experimental::binder::SecurityPolicy>
          security_policy,
      std::function<void()> on_destruct_callback = nullptr);
  ~WireReaderImpl() override;
 
  void Orphan() override { Unref(); }
 
  std::shared_ptr<WireWriter> SetupTransport(
      std::unique_ptr<Binder> binder) override;
 
  absl::Status ProcessTransaction(transaction_code_t code,
                                  ReadableParcel* parcel, int uid);
 
  void SendSetupTransport(Binder* binder);
 
  // TODO(waynetu): In the testing environment, we still use this method (on
  // another WireReader instance) for server-side transport setup, and thus it
  // is marked as public. Try moving this method back to private, and hopefully
  // we can also avoid moving |other_end_binder_| out in the implementation.
  std::unique_ptr<Binder> RecvSetupTransport();
 
 private:
  absl::Status ProcessStreamingTransaction(transaction_code_t code,
                                           ReadableParcel* parcel);
  absl::Status ProcessStreamingTransactionImpl(transaction_code_t code,
                                               ReadableParcel* parcel,
                                               int* cancellation_flags)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
  std::shared_ptr<TransportStreamReceiver> transport_stream_receiver_;
  absl::Notification connection_noti_;
  grpc_core::Mutex mu_;
  bool connected_ ABSL_GUARDED_BY(mu_) = false;
  bool recvd_setup_transport_ ABSL_GUARDED_BY(mu_) = false;
  // NOTE: other_end_binder_ will be moved out when RecvSetupTransport() is
  // called. Be cautious not to access it afterward.
  std::unique_ptr<Binder> other_end_binder_;
  absl::flat_hash_map<transaction_code_t, int32_t> expected_seq_num_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<transaction_code_t, std::string> message_buffer_
      ABSL_GUARDED_BY(mu_);
  std::unique_ptr<TransactionReceiver> tx_receiver_;
  bool is_client_;
  std::shared_ptr<grpc::experimental::binder::SecurityPolicy> security_policy_;
  // When WireReaderImpl gets destructed, call on_destruct_callback_. This is
  // mostly for decrementing the reference count of its transport.
  std::function<void()> on_destruct_callback_;
 
  // ACK every 16k bytes.
  static constexpr int64_t kFlowControlAckBytes = 16 * 1024;
  int64_t num_incoming_bytes_ ABSL_GUARDED_BY(mu_) = 0;
  int64_t num_acknowledged_bytes_ ABSL_GUARDED_BY(mu_) = 0;
 
  // Used to send ACK.
  std::shared_ptr<WireWriter> wire_writer_;
};
 
}  // namespace grpc_binder
 
#endif  // GRPC_CORE_EXT_TRANSPORT_BINDER_WIRE_FORMAT_WIRE_READER_IMPL_H