|
| 1 | +/* |
| 2 | + * Copyright 2018 Google |
| 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 FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_STREAM_H_ |
| 18 | +#define FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_STREAM_H_ |
| 19 | + |
| 20 | +#include <memory> |
| 21 | +#include <string> |
| 22 | + |
| 23 | +#include "Firestore/core/src/firebase/firestore/auth/credentials_provider.h" |
| 24 | +#include "Firestore/core/src/firebase/firestore/auth/token.h" |
| 25 | +#include "Firestore/core/src/firebase/firestore/remote/datastore.h" |
| 26 | +#include "Firestore/core/src/firebase/firestore/remote/exponential_backoff.h" |
| 27 | +#include "Firestore/core/src/firebase/firestore/remote/grpc_completion.h" |
| 28 | +#include "Firestore/core/src/firebase/firestore/remote/grpc_stream.h" |
| 29 | +#include "Firestore/core/src/firebase/firestore/remote/stream_objc_bridge.h" |
| 30 | +#include "Firestore/core/src/firebase/firestore/util/async_queue.h" |
| 31 | +#include "Firestore/core/src/firebase/firestore/util/status.h" |
| 32 | +#include "Firestore/core/src/firebase/firestore/util/statusor.h" |
| 33 | +#include "absl/strings/string_view.h" |
| 34 | +#include "grpcpp/support/byte_buffer.h" |
| 35 | + |
| 36 | +namespace firebase { |
| 37 | +namespace firestore { |
| 38 | +namespace remote { |
| 39 | + |
| 40 | +/** |
| 41 | + * A `Stream` is an abstract base class that represents a bidirectional |
| 42 | + * streaming connection to the Firestore backend. It's built on top of gRPC C++ |
| 43 | + * library and adds several critical features for our clients: |
| 44 | + * |
| 45 | + * - Exponential backoff on failure (independent of the gRPC mechanism) |
| 46 | + * - Authentication via CredentialsProvider |
| 47 | + * - Dispatching all callbacks into the shared Firestore async queue |
| 48 | + * - Closing idle streams after 60 seconds of inactivity |
| 49 | + * |
| 50 | + * Subclasses of `Stream`: |
| 51 | + * |
| 52 | + * - Implement serialization and deserialization of protocol buffers |
| 53 | + * - Notify their delegate about stream open/read/error events |
| 54 | + * - Create and finish the underlying gRPC streams. |
| 55 | + * |
| 56 | + * ## Starting and Stopping |
| 57 | + * |
| 58 | + * Streams are stateful and need to be `Start`ed before messages can |
| 59 | + * be sent and received. A `Stream` can be started and stopped repeatedly. |
| 60 | + */ |
| 61 | +class Stream : public GrpcStreamObserver, |
| 62 | + public std::enable_shared_from_this<Stream> { |
| 63 | + public: |
| 64 | + /** |
| 65 | + * `Stream` can be in one of 5 states (each described in detail below) |
| 66 | + * shown in the following state transition diagram: |
| 67 | + * |
| 68 | + * Start() called auth & connection succeeded |
| 69 | + * INITIAL ----------------> STARTING -----------------------------> OPEN |
| 70 | + * ^ | | |
| 71 | + * | | error occurred | |
| 72 | + * | \-----------------------------v-----/ |
| 73 | + * | | |
| 74 | + * backoff | | |
| 75 | + * elapsed | Start() called | |
| 76 | + * \--- BACKOFF <---------------- ERROR |
| 77 | + * |
| 78 | + * [any state] --------------------------> INITIAL |
| 79 | + * Stop() called or |
| 80 | + * idle timer expired |
| 81 | + */ |
| 82 | + enum class State { |
| 83 | + /** |
| 84 | + * The stream is not yet running and there's no error condition. |
| 85 | + * Calling `Start` will start the stream immediately without backoff. |
| 86 | + * While in this state, `IsStarted` will return false. |
| 87 | + */ |
| 88 | + Initial, |
| 89 | + |
| 90 | + /** |
| 91 | + * The stream is starting, either waiting for an auth token or for the |
| 92 | + * stream to successfully open. While in this state, `IsStarted` will |
| 93 | + * return true but `IsOpen` will return false. |
| 94 | + */ |
| 95 | + Starting, |
| 96 | + |
| 97 | + /** |
| 98 | + * The stream is up and running. Requests and responses can flow |
| 99 | + * freely. Both `IsStarted` and `IsOpen` will return true. |
| 100 | + */ |
| 101 | + Open, |
| 102 | + |
| 103 | + /** |
| 104 | + * The stream encountered an error. The next start attempt will back off. |
| 105 | + * While in this state, `IsStarted` will return false. |
| 106 | + */ |
| 107 | + Error, |
| 108 | + |
| 109 | + /** |
| 110 | + * An in-between state after an error where the stream is waiting before |
| 111 | + * re-starting. After waiting is complete, the stream will try to open. |
| 112 | + * While in this state, `IsStarted` will return true but `IsOpen` will |
| 113 | + * return false. |
| 114 | + */ |
| 115 | + Backoff |
| 116 | + }; |
| 117 | + |
| 118 | + Stream(util::AsyncQueue* async_queue, |
| 119 | + auth::CredentialsProvider* credentials_provider, |
| 120 | + Datastore* datastore, |
| 121 | + util::TimerId backoff_timer_id, |
| 122 | + util::TimerId idle_timer_id); |
| 123 | + |
| 124 | + /** |
| 125 | + * Starts the stream. Only allowed if `IsStarted` returns false. The stream is |
| 126 | + * not immediately ready for use: `OnStreamStart` will be invoked when the |
| 127 | + * stream is ready for outbound requests, at which point `IsOpen` will return |
| 128 | + * true. |
| 129 | + * |
| 130 | + * When start returns, `IsStarted` will return true. |
| 131 | + */ |
| 132 | + void Start(); |
| 133 | + |
| 134 | + /** |
| 135 | + * Stops the stream. This call is idempotent and allowed regardless of the |
| 136 | + * current `IsStarted` state. |
| 137 | + * |
| 138 | + * When stop returns, `IsStarted` and `IsOpen` will both return false. |
| 139 | + */ |
| 140 | + void Stop(); |
| 141 | + |
| 142 | + /** |
| 143 | + * Returns true if `Start` has been called and no error has occurred. True |
| 144 | + * indicates the stream is open or in the process of opening (which |
| 145 | + * encompasses respecting backoff, getting auth tokens, and starting the |
| 146 | + * actual stream). Use `IsOpen` to determine if the stream is open and ready |
| 147 | + * for outbound requests. |
| 148 | + */ |
| 149 | + bool IsStarted() const; |
| 150 | + |
| 151 | + /** |
| 152 | + * Returns true if the underlying stream is open (`OnStreamStart` has been |
| 153 | + * called) and the stream is ready for outbound requests. |
| 154 | + */ |
| 155 | + bool IsOpen() const; |
| 156 | + |
| 157 | + /** |
| 158 | + * After an error, the stream will usually back off on the next attempt to |
| 159 | + * start it. If the error warrants an immediate restart of the stream, the |
| 160 | + * caller can use this to indicate that the stream should not back off. |
| 161 | + * |
| 162 | + * Each error will call `OnStreamClose`. That function can decide to |
| 163 | + * cancel backoff if required. |
| 164 | + */ |
| 165 | + void InhibitBackoff(); |
| 166 | + |
| 167 | + /** |
| 168 | + * Marks this stream as idle. If no further actions are performed on the |
| 169 | + * stream for one minute, the stream will automatically close itself and |
| 170 | + * notify the stream's `OnClose` handler with Status::OK. The stream will then |
| 171 | + * be in a non-started state, requiring the caller to start the stream again |
| 172 | + * before further use. |
| 173 | + * |
| 174 | + * Only streams that are in state 'Open' can be marked idle, as all other |
| 175 | + * states imply pending network operations. |
| 176 | + */ |
| 177 | + void MarkIdle(); |
| 178 | + |
| 179 | + /** |
| 180 | + * Marks the stream as active again, preventing auto-closing of the stream. |
| 181 | + * Can be called from any state -- if the stream is not in state `Open`, this |
| 182 | + * is a no-op. |
| 183 | + */ |
| 184 | + void CancelIdleCheck(); |
| 185 | + |
| 186 | + // `GrpcStreamObserver` interface -- do not use. |
| 187 | + void OnStreamStart() override; |
| 188 | + void OnStreamRead(const grpc::ByteBuffer& message) override; |
| 189 | + void OnStreamError(const util::Status& status) override; |
| 190 | + |
| 191 | + protected: |
| 192 | + // `Stream` expects all its methods to be called on the worker queue. |
| 193 | + void EnsureOnQueue() const; |
| 194 | + void Write(grpc::ByteBuffer&& message); |
| 195 | + std::string GetDebugDescription() const; |
| 196 | + |
| 197 | + ExponentialBackoff backoff_; |
| 198 | + |
| 199 | + private: |
| 200 | + // The interface for the derived classes. |
| 201 | + |
| 202 | + virtual std::unique_ptr<GrpcStream> CreateGrpcStream( |
| 203 | + Datastore* datastore, absl::string_view token) = 0; |
| 204 | + virtual void TearDown(GrpcStream* stream) = 0; |
| 205 | + virtual void NotifyStreamOpen() = 0; |
| 206 | + virtual util::Status NotifyStreamResponse( |
| 207 | + const grpc::ByteBuffer& message) = 0; |
| 208 | + virtual void NotifyStreamClose(const util::Status& status) = 0; |
| 209 | + // PORTING NOTE: C++ cannot rely on RTTI, unlike other platforms. |
| 210 | + virtual std::string GetDebugName() const = 0; |
| 211 | + |
| 212 | + void Close(const util::Status& status); |
| 213 | + void HandleErrorStatus(const util::Status& status); |
| 214 | + |
| 215 | + void RequestCredentials(); |
| 216 | + void ResumeStartWithCredentials( |
| 217 | + const util::StatusOr<auth::Token>& maybe_token); |
| 218 | + |
| 219 | + void BackoffAndTryRestarting(); |
| 220 | + void StopDueToIdleness(); |
| 221 | + |
| 222 | + State state_ = State::Initial; |
| 223 | + |
| 224 | + std::unique_ptr<GrpcStream> grpc_stream_; |
| 225 | + |
| 226 | + auth::CredentialsProvider* credentials_provider_ = nullptr; |
| 227 | + util::AsyncQueue* worker_queue_ = nullptr; |
| 228 | + Datastore* datastore_ = nullptr; |
| 229 | + |
| 230 | + util::TimerId idle_timer_id_{}; |
| 231 | + util::DelayedOperation idleness_timer_; |
| 232 | + |
| 233 | + // Used to prevent auth if the stream happens to be restarted before token is |
| 234 | + // received. |
| 235 | + int close_count_ = 0; |
| 236 | +}; |
| 237 | + |
| 238 | +} // namespace remote |
| 239 | +} // namespace firestore |
| 240 | +} // namespace firebase |
| 241 | + |
| 242 | +#endif // FIRESTORE_CORE_SRC_FIREBASE_FIRESTORE_REMOTE_STREAM_H_ |
0 commit comments