1 /* 2 * Copyright 2004 The WebRTC Project Authors. All rights reserved. 3 * 4 * Use of this source code is governed by a BSD-style license 5 * that can be found in the LICENSE file in the root of the source 6 * tree. An additional intellectual property rights grant can be found 7 * in the file PATENTS. All contributing project authors may 8 * be found in the AUTHORS file in the root of the source tree. 9 */ 10 11 #ifndef WEBRTC_BASE_STREAM_H_ 12 #define WEBRTC_BASE_STREAM_H_ 13 14 #include <stdio.h> 15 16 #include "webrtc/base/basictypes.h" 17 #include "webrtc/base/buffer.h" 18 #include "webrtc/base/criticalsection.h" 19 #include "webrtc/base/logging.h" 20 #include "webrtc/base/messagehandler.h" 21 #include "webrtc/base/messagequeue.h" 22 #include "webrtc/base/scoped_ptr.h" 23 #include "webrtc/base/sigslot.h" 24 25 namespace rtc { 26 27 /////////////////////////////////////////////////////////////////////////////// 28 // StreamInterface is a generic asynchronous stream interface, supporting read, 29 // write, and close operations, and asynchronous signalling of state changes. 30 // The interface is designed with file, memory, and socket implementations in 31 // mind. Some implementations offer extended operations, such as seeking. 32 /////////////////////////////////////////////////////////////////////////////// 33 34 // The following enumerations are declared outside of the StreamInterface 35 // class for brevity in use. 36 37 // The SS_OPENING state indicates that the stream will signal open or closed 38 // in the future. 39 enum StreamState { SS_CLOSED, SS_OPENING, SS_OPEN }; 40 41 // Stream read/write methods return this value to indicate various success 42 // and failure conditions described below. 43 enum StreamResult { SR_ERROR, SR_SUCCESS, SR_BLOCK, SR_EOS }; 44 45 // StreamEvents are used to asynchronously signal state transitionss. The flags 46 // may be combined. 47 // SE_OPEN: The stream has transitioned to the SS_OPEN state 48 // SE_CLOSE: The stream has transitioned to the SS_CLOSED state 49 // SE_READ: Data is available, so Read is likely to not return SR_BLOCK 50 // SE_WRITE: Data can be written, so Write is likely to not return SR_BLOCK 51 enum StreamEvent { SE_OPEN = 1, SE_READ = 2, SE_WRITE = 4, SE_CLOSE = 8 }; 52 53 class Thread; 54 55 struct StreamEventData : public MessageData { 56 int events, error; StreamEventDataStreamEventData57 StreamEventData(int ev, int er) : events(ev), error(er) { } 58 }; 59 60 class StreamInterface : public MessageHandler { 61 public: 62 enum { 63 MSG_POST_EVENT = 0xF1F1, MSG_MAX = MSG_POST_EVENT 64 }; 65 66 ~StreamInterface() override; 67 68 virtual StreamState GetState() const = 0; 69 70 // Read attempts to fill buffer of size buffer_len. Write attempts to send 71 // data_len bytes stored in data. The variables read and write are set only 72 // on SR_SUCCESS (see below). Likewise, error is only set on SR_ERROR. 73 // Read and Write return a value indicating: 74 // SR_ERROR: an error occurred, which is returned in a non-null error 75 // argument. Interpretation of the error requires knowledge of the 76 // stream's concrete type, which limits its usefulness. 77 // SR_SUCCESS: some number of bytes were successfully written, which is 78 // returned in a non-null read/write argument. 79 // SR_BLOCK: the stream is in non-blocking mode, and the operation would 80 // block, or the stream is in SS_OPENING state. 81 // SR_EOS: the end-of-stream has been reached, or the stream is in the 82 // SS_CLOSED state. 83 virtual StreamResult Read(void* buffer, size_t buffer_len, 84 size_t* read, int* error) = 0; 85 virtual StreamResult Write(const void* data, size_t data_len, 86 size_t* written, int* error) = 0; 87 // Attempt to transition to the SS_CLOSED state. SE_CLOSE will not be 88 // signalled as a result of this call. 89 virtual void Close() = 0; 90 91 // Streams may signal one or more StreamEvents to indicate state changes. 92 // The first argument identifies the stream on which the state change occured. 93 // The second argument is a bit-wise combination of StreamEvents. 94 // If SE_CLOSE is signalled, then the third argument is the associated error 95 // code. Otherwise, the value is undefined. 96 // Note: Not all streams will support asynchronous event signalling. However, 97 // SS_OPENING and SR_BLOCK returned from stream member functions imply that 98 // certain events will be raised in the future. 99 sigslot::signal3<StreamInterface*, int, int> SignalEvent; 100 101 // Like calling SignalEvent, but posts a message to the specified thread, 102 // which will call SignalEvent. This helps unroll the stack and prevent 103 // re-entrancy. 104 void PostEvent(Thread* t, int events, int err); 105 // Like the aforementioned method, but posts to the current thread. 106 void PostEvent(int events, int err); 107 108 // 109 // OPTIONAL OPERATIONS 110 // 111 // Not all implementations will support the following operations. In general, 112 // a stream will only support an operation if it reasonably efficient to do 113 // so. For example, while a socket could buffer incoming data to support 114 // seeking, it will not do so. Instead, a buffering stream adapter should 115 // be used. 116 // 117 // Even though several of these operations are related, you should 118 // always use whichever operation is most relevant. For example, you may 119 // be tempted to use GetSize() and GetPosition() to deduce the result of 120 // GetAvailable(). However, a stream which is read-once may support the 121 // latter operation but not the former. 122 // 123 124 // The following four methods are used to avoid copying data multiple times. 125 126 // GetReadData returns a pointer to a buffer which is owned by the stream. 127 // The buffer contains data_len bytes. NULL is returned if no data is 128 // available, or if the method fails. If the caller processes the data, it 129 // must call ConsumeReadData with the number of processed bytes. GetReadData 130 // does not require a matching call to ConsumeReadData if the data is not 131 // processed. Read and ConsumeReadData invalidate the buffer returned by 132 // GetReadData. 133 virtual const void* GetReadData(size_t* data_len); ConsumeReadData(size_t used)134 virtual void ConsumeReadData(size_t used) {} 135 136 // GetWriteBuffer returns a pointer to a buffer which is owned by the stream. 137 // The buffer has a capacity of buf_len bytes. NULL is returned if there is 138 // no buffer available, or if the method fails. The call may write data to 139 // the buffer, and then call ConsumeWriteBuffer with the number of bytes 140 // written. GetWriteBuffer does not require a matching call to 141 // ConsumeWriteData if no data is written. Write, ForceWrite, and 142 // ConsumeWriteData invalidate the buffer returned by GetWriteBuffer. 143 // TODO: Allow the caller to specify a minimum buffer size. If the specified 144 // amount of buffer is not yet available, return NULL and Signal SE_WRITE 145 // when it is available. If the requested amount is too large, return an 146 // error. 147 virtual void* GetWriteBuffer(size_t* buf_len); ConsumeWriteBuffer(size_t used)148 virtual void ConsumeWriteBuffer(size_t used) {} 149 150 // Write data_len bytes found in data, circumventing any throttling which 151 // would could cause SR_BLOCK to be returned. Returns true if all the data 152 // was written. Otherwise, the method is unsupported, or an unrecoverable 153 // error occurred, and the error value is set. This method should be used 154 // sparingly to write critical data which should not be throttled. A stream 155 // which cannot circumvent its blocking constraints should not implement this 156 // method. 157 // NOTE: This interface is being considered experimentally at the moment. It 158 // would be used by JUDP and BandwidthStream as a way to circumvent certain 159 // soft limits in writing. 160 //virtual bool ForceWrite(const void* data, size_t data_len, int* error) { 161 // if (error) *error = -1; 162 // return false; 163 //} 164 165 // Seek to a byte offset from the beginning of the stream. Returns false if 166 // the stream does not support seeking, or cannot seek to the specified 167 // position. 168 virtual bool SetPosition(size_t position); 169 170 // Get the byte offset of the current position from the start of the stream. 171 // Returns false if the position is not known. 172 virtual bool GetPosition(size_t* position) const; 173 174 // Get the byte length of the entire stream. Returns false if the length 175 // is not known. 176 virtual bool GetSize(size_t* size) const; 177 178 // Return the number of Read()-able bytes remaining before end-of-stream. 179 // Returns false if not known. 180 virtual bool GetAvailable(size_t* size) const; 181 182 // Return the number of Write()-able bytes remaining before end-of-stream. 183 // Returns false if not known. 184 virtual bool GetWriteRemaining(size_t* size) const; 185 186 // Return true if flush is successful. 187 virtual bool Flush(); 188 189 // Communicates the amount of data which will be written to the stream. The 190 // stream may choose to preallocate memory to accomodate this data. The 191 // stream may return false to indicate that there is not enough room (ie, 192 // Write will return SR_EOS/SR_ERROR at some point). Note that calling this 193 // function should not affect the existing state of data in the stream. 194 virtual bool ReserveSize(size_t size); 195 196 // 197 // CONVENIENCE METHODS 198 // 199 // These methods are implemented in terms of other methods, for convenience. 200 // 201 202 // Seek to the start of the stream. Rewind()203 inline bool Rewind() { return SetPosition(0); } 204 205 // WriteAll is a helper function which repeatedly calls Write until all the 206 // data is written, or something other than SR_SUCCESS is returned. Note that 207 // unlike Write, the argument 'written' is always set, and may be non-zero 208 // on results other than SR_SUCCESS. The remaining arguments have the 209 // same semantics as Write. 210 StreamResult WriteAll(const void* data, size_t data_len, 211 size_t* written, int* error); 212 213 // Similar to ReadAll. Calls Read until buffer_len bytes have been read, or 214 // until a non-SR_SUCCESS result is returned. 'read' is always set. 215 StreamResult ReadAll(void* buffer, size_t buffer_len, 216 size_t* read, int* error); 217 218 // ReadLine is a helper function which repeatedly calls Read until it hits 219 // the end-of-line character, or something other than SR_SUCCESS. 220 // TODO: this is too inefficient to keep here. Break this out into a buffered 221 // readline object or adapter 222 StreamResult ReadLine(std::string* line); 223 224 protected: 225 StreamInterface(); 226 227 // MessageHandler Interface 228 void OnMessage(Message* msg) override; 229 230 private: 231 RTC_DISALLOW_COPY_AND_ASSIGN(StreamInterface); 232 }; 233 234 /////////////////////////////////////////////////////////////////////////////// 235 // StreamAdapterInterface is a convenient base-class for adapting a stream. 236 // By default, all operations are pass-through. Override the methods that you 237 // require adaptation. Streams should really be upgraded to reference-counted. 238 // In the meantime, use the owned flag to indicate whether the adapter should 239 // own the adapted stream. 240 /////////////////////////////////////////////////////////////////////////////// 241 242 class StreamAdapterInterface : public StreamInterface, 243 public sigslot::has_slots<> { 244 public: 245 explicit StreamAdapterInterface(StreamInterface* stream, bool owned = true); 246 247 // Core Stream Interface 248 StreamState GetState() const override; 249 StreamResult Read(void* buffer, 250 size_t buffer_len, 251 size_t* read, 252 int* error) override; 253 StreamResult Write(const void* data, 254 size_t data_len, 255 size_t* written, 256 int* error) override; 257 void Close() override; 258 259 // Optional Stream Interface 260 /* Note: Many stream adapters were implemented prior to this Read/Write 261 interface. Therefore, a simple pass through of data in those cases may 262 be broken. At a later time, we should do a once-over pass of all 263 adapters, and make them compliant with these interfaces, after which this 264 code can be uncommented. 265 virtual const void* GetReadData(size_t* data_len) { 266 return stream_->GetReadData(data_len); 267 } 268 virtual void ConsumeReadData(size_t used) { 269 stream_->ConsumeReadData(used); 270 } 271 272 virtual void* GetWriteBuffer(size_t* buf_len) { 273 return stream_->GetWriteBuffer(buf_len); 274 } 275 virtual void ConsumeWriteBuffer(size_t used) { 276 stream_->ConsumeWriteBuffer(used); 277 } 278 */ 279 280 /* Note: This interface is currently undergoing evaluation. 281 virtual bool ForceWrite(const void* data, size_t data_len, int* error) { 282 return stream_->ForceWrite(data, data_len, error); 283 } 284 */ 285 286 bool SetPosition(size_t position) override; 287 bool GetPosition(size_t* position) const override; 288 bool GetSize(size_t* size) const override; 289 bool GetAvailable(size_t* size) const override; 290 bool GetWriteRemaining(size_t* size) const override; 291 bool ReserveSize(size_t size) override; 292 bool Flush() override; 293 294 void Attach(StreamInterface* stream, bool owned = true); 295 StreamInterface* Detach(); 296 297 protected: 298 ~StreamAdapterInterface() override; 299 300 // Note that the adapter presents itself as the origin of the stream events, 301 // since users of the adapter may not recognize the adapted object. 302 virtual void OnEvent(StreamInterface* stream, int events, int err); stream()303 StreamInterface* stream() { return stream_; } 304 305 private: 306 StreamInterface* stream_; 307 bool owned_; 308 RTC_DISALLOW_COPY_AND_ASSIGN(StreamAdapterInterface); 309 }; 310 311 /////////////////////////////////////////////////////////////////////////////// 312 // StreamTap is a non-modifying, pass-through adapter, which copies all data 313 // in either direction to the tap. Note that errors or blocking on writing to 314 // the tap will prevent further tap writes from occurring. 315 /////////////////////////////////////////////////////////////////////////////// 316 317 class StreamTap : public StreamAdapterInterface { 318 public: 319 explicit StreamTap(StreamInterface* stream, StreamInterface* tap); 320 ~StreamTap() override; 321 322 void AttachTap(StreamInterface* tap); 323 StreamInterface* DetachTap(); 324 StreamResult GetTapResult(int* error); 325 326 // StreamAdapterInterface Interface 327 StreamResult Read(void* buffer, 328 size_t buffer_len, 329 size_t* read, 330 int* error) override; 331 StreamResult Write(const void* data, 332 size_t data_len, 333 size_t* written, 334 int* error) override; 335 336 private: 337 scoped_ptr<StreamInterface> tap_; 338 StreamResult tap_result_; 339 int tap_error_; 340 RTC_DISALLOW_COPY_AND_ASSIGN(StreamTap); 341 }; 342 343 /////////////////////////////////////////////////////////////////////////////// 344 // NullStream gives errors on read, and silently discards all written data. 345 /////////////////////////////////////////////////////////////////////////////// 346 347 class NullStream : public StreamInterface { 348 public: 349 NullStream(); 350 ~NullStream() override; 351 352 // StreamInterface Interface 353 StreamState GetState() const override; 354 StreamResult Read(void* buffer, 355 size_t buffer_len, 356 size_t* read, 357 int* error) override; 358 StreamResult Write(const void* data, 359 size_t data_len, 360 size_t* written, 361 int* error) override; 362 void Close() override; 363 }; 364 365 /////////////////////////////////////////////////////////////////////////////// 366 // FileStream is a simple implementation of a StreamInterface, which does not 367 // support asynchronous notification. 368 /////////////////////////////////////////////////////////////////////////////// 369 370 class FileStream : public StreamInterface { 371 public: 372 FileStream(); 373 ~FileStream() override; 374 375 // The semantics of filename and mode are the same as stdio's fopen 376 virtual bool Open(const std::string& filename, const char* mode, int* error); 377 virtual bool OpenShare(const std::string& filename, const char* mode, 378 int shflag, int* error); 379 380 // By default, reads and writes are buffered for efficiency. Disabling 381 // buffering causes writes to block until the bytes on disk are updated. 382 virtual bool DisableBuffering(); 383 384 StreamState GetState() const override; 385 StreamResult Read(void* buffer, 386 size_t buffer_len, 387 size_t* read, 388 int* error) override; 389 StreamResult Write(const void* data, 390 size_t data_len, 391 size_t* written, 392 int* error) override; 393 void Close() override; 394 bool SetPosition(size_t position) override; 395 bool GetPosition(size_t* position) const override; 396 bool GetSize(size_t* size) const override; 397 bool GetAvailable(size_t* size) const override; 398 bool ReserveSize(size_t size) override; 399 400 bool Flush() override; 401 402 #if defined(WEBRTC_POSIX) && !defined(__native_client__) 403 // Tries to aquire an exclusive lock on the file. 404 // Use OpenShare(...) on win32 to get similar functionality. 405 bool TryLock(); 406 bool Unlock(); 407 #endif 408 409 // Note: Deprecated in favor of Filesystem::GetFileSize(). 410 static bool GetSize(const std::string& filename, size_t* size); 411 412 protected: 413 virtual void DoClose(); 414 415 FILE* file_; 416 417 private: 418 RTC_DISALLOW_COPY_AND_ASSIGN(FileStream); 419 }; 420 421 /////////////////////////////////////////////////////////////////////////////// 422 // MemoryStream is a simple implementation of a StreamInterface over in-memory 423 // data. Data is read and written at the current seek position. Reads return 424 // end-of-stream when they reach the end of data. Writes actually extend the 425 // end of data mark. 426 /////////////////////////////////////////////////////////////////////////////// 427 428 class MemoryStreamBase : public StreamInterface { 429 public: 430 StreamState GetState() const override; 431 StreamResult Read(void* buffer, 432 size_t bytes, 433 size_t* bytes_read, 434 int* error) override; 435 StreamResult Write(const void* buffer, 436 size_t bytes, 437 size_t* bytes_written, 438 int* error) override; 439 void Close() override; 440 bool SetPosition(size_t position) override; 441 bool GetPosition(size_t* position) const override; 442 bool GetSize(size_t* size) const override; 443 bool GetAvailable(size_t* size) const override; 444 bool ReserveSize(size_t size) override; 445 GetBuffer()446 char* GetBuffer() { return buffer_; } GetBuffer()447 const char* GetBuffer() const { return buffer_; } 448 449 protected: 450 MemoryStreamBase(); 451 452 virtual StreamResult DoReserve(size_t size, int* error); 453 454 // Invariant: 0 <= seek_position <= data_length_ <= buffer_length_ 455 char* buffer_; 456 size_t buffer_length_; 457 size_t data_length_; 458 size_t seek_position_; 459 460 private: 461 RTC_DISALLOW_COPY_AND_ASSIGN(MemoryStreamBase); 462 }; 463 464 // MemoryStream dynamically resizes to accomodate written data. 465 466 class MemoryStream : public MemoryStreamBase { 467 public: 468 MemoryStream(); 469 explicit MemoryStream(const char* data); // Calls SetData(data, strlen(data)) 470 MemoryStream(const void* data, size_t length); // Calls SetData(data, length) 471 ~MemoryStream() override; 472 473 void SetData(const void* data, size_t length); 474 475 protected: 476 StreamResult DoReserve(size_t size, int* error) override; 477 // Memory Streams are aligned for efficiency. 478 static const int kAlignment = 16; 479 char* buffer_alloc_; 480 }; 481 482 // ExternalMemoryStream adapts an external memory buffer, so writes which would 483 // extend past the end of the buffer will return end-of-stream. 484 485 class ExternalMemoryStream : public MemoryStreamBase { 486 public: 487 ExternalMemoryStream(); 488 ExternalMemoryStream(void* data, size_t length); 489 ~ExternalMemoryStream() override; 490 491 void SetData(void* data, size_t length); 492 }; 493 494 // FifoBuffer allows for efficient, thread-safe buffering of data between 495 // writer and reader. As the data can wrap around the end of the buffer, 496 // MemoryStreamBase can't help us here. 497 498 class FifoBuffer : public StreamInterface { 499 public: 500 // Creates a FIFO buffer with the specified capacity. 501 explicit FifoBuffer(size_t length); 502 // Creates a FIFO buffer with the specified capacity and owner 503 FifoBuffer(size_t length, Thread* owner); 504 ~FifoBuffer() override; 505 // Gets the amount of data currently readable from the buffer. 506 bool GetBuffered(size_t* data_len) const; 507 // Resizes the buffer to the specified capacity. Fails if data_length_ > size 508 bool SetCapacity(size_t length); 509 510 // Read into |buffer| with an offset from the current read position, offset 511 // is specified in number of bytes. 512 // This method doesn't adjust read position nor the number of available 513 // bytes, user has to call ConsumeReadData() to do this. 514 StreamResult ReadOffset(void* buffer, size_t bytes, size_t offset, 515 size_t* bytes_read); 516 517 // Write |buffer| with an offset from the current write position, offset is 518 // specified in number of bytes. 519 // This method doesn't adjust the number of buffered bytes, user has to call 520 // ConsumeWriteBuffer() to do this. 521 StreamResult WriteOffset(const void* buffer, size_t bytes, size_t offset, 522 size_t* bytes_written); 523 524 // StreamInterface methods 525 StreamState GetState() const override; 526 StreamResult Read(void* buffer, 527 size_t bytes, 528 size_t* bytes_read, 529 int* error) override; 530 StreamResult Write(const void* buffer, 531 size_t bytes, 532 size_t* bytes_written, 533 int* error) override; 534 void Close() override; 535 const void* GetReadData(size_t* data_len) override; 536 void ConsumeReadData(size_t used) override; 537 void* GetWriteBuffer(size_t* buf_len) override; 538 void ConsumeWriteBuffer(size_t used) override; 539 bool GetWriteRemaining(size_t* size) const override; 540 541 private: 542 // Helper method that implements ReadOffset. Caller must acquire a lock 543 // when calling this method. 544 StreamResult ReadOffsetLocked(void* buffer, size_t bytes, size_t offset, 545 size_t* bytes_read); 546 547 // Helper method that implements WriteOffset. Caller must acquire a lock 548 // when calling this method. 549 StreamResult WriteOffsetLocked(const void* buffer, size_t bytes, 550 size_t offset, size_t* bytes_written); 551 552 StreamState state_; // keeps the opened/closed state of the stream 553 scoped_ptr<char[]> buffer_; // the allocated buffer 554 size_t buffer_length_; // size of the allocated buffer 555 size_t data_length_; // amount of readable data in the buffer 556 size_t read_position_; // offset to the readable data 557 Thread* owner_; // stream callbacks are dispatched on this thread 558 mutable CriticalSection crit_; // object lock 559 RTC_DISALLOW_COPY_AND_ASSIGN(FifoBuffer); 560 }; 561 562 /////////////////////////////////////////////////////////////////////////////// 563 564 class LoggingAdapter : public StreamAdapterInterface { 565 public: 566 LoggingAdapter(StreamInterface* stream, LoggingSeverity level, 567 const std::string& label, bool hex_mode = false); 568 569 void set_label(const std::string& label); 570 571 StreamResult Read(void* buffer, 572 size_t buffer_len, 573 size_t* read, 574 int* error) override; 575 StreamResult Write(const void* data, 576 size_t data_len, 577 size_t* written, 578 int* error) override; 579 void Close() override; 580 581 protected: 582 void OnEvent(StreamInterface* stream, int events, int err) override; 583 584 private: 585 LoggingSeverity level_; 586 std::string label_; 587 bool hex_mode_; 588 LogMultilineState lms_; 589 590 RTC_DISALLOW_COPY_AND_ASSIGN(LoggingAdapter); 591 }; 592 593 /////////////////////////////////////////////////////////////////////////////// 594 // StringStream - Reads/Writes to an external std::string 595 /////////////////////////////////////////////////////////////////////////////// 596 597 class StringStream : public StreamInterface { 598 public: 599 explicit StringStream(std::string* str); 600 explicit StringStream(const std::string& str); 601 602 StreamState GetState() const override; 603 StreamResult Read(void* buffer, 604 size_t buffer_len, 605 size_t* read, 606 int* error) override; 607 StreamResult Write(const void* data, 608 size_t data_len, 609 size_t* written, 610 int* error) override; 611 void Close() override; 612 bool SetPosition(size_t position) override; 613 bool GetPosition(size_t* position) const override; 614 bool GetSize(size_t* size) const override; 615 bool GetAvailable(size_t* size) const override; 616 bool ReserveSize(size_t size) override; 617 618 private: 619 std::string& str_; 620 size_t read_pos_; 621 bool read_only_; 622 }; 623 624 /////////////////////////////////////////////////////////////////////////////// 625 // StreamReference - A reference counting stream adapter 626 /////////////////////////////////////////////////////////////////////////////// 627 628 // Keep in mind that the streams and adapters defined in this file are 629 // not thread-safe, so this has limited uses. 630 631 // A StreamRefCount holds the reference count and a pointer to the 632 // wrapped stream. It deletes the wrapped stream when there are no 633 // more references. We can then have multiple StreamReference 634 // instances pointing to one StreamRefCount, all wrapping the same 635 // stream. 636 637 class StreamReference : public StreamAdapterInterface { 638 class StreamRefCount; 639 public: 640 // Constructor for the first reference to a stream 641 // Note: get more references through NewReference(). Use this 642 // constructor only once on a given stream. 643 explicit StreamReference(StreamInterface* stream); GetStream()644 StreamInterface* GetStream() { return stream(); } 645 StreamInterface* NewReference(); 646 ~StreamReference() override; 647 648 private: 649 class StreamRefCount { 650 public: StreamRefCount(StreamInterface * stream)651 explicit StreamRefCount(StreamInterface* stream) 652 : stream_(stream), ref_count_(1) { 653 } AddReference()654 void AddReference() { 655 CritScope lock(&cs_); 656 ++ref_count_; 657 } Release()658 void Release() { 659 int ref_count; 660 { // Atomic ops would have been a better fit here. 661 CritScope lock(&cs_); 662 ref_count = --ref_count_; 663 } 664 if (ref_count == 0) { 665 delete stream_; 666 delete this; 667 } 668 } 669 private: 670 StreamInterface* stream_; 671 int ref_count_; 672 CriticalSection cs_; 673 RTC_DISALLOW_COPY_AND_ASSIGN(StreamRefCount); 674 }; 675 676 // Constructor for adding references 677 explicit StreamReference(StreamRefCount* stream_ref_count, 678 StreamInterface* stream); 679 680 StreamRefCount* stream_ref_count_; 681 RTC_DISALLOW_COPY_AND_ASSIGN(StreamReference); 682 }; 683 684 /////////////////////////////////////////////////////////////////////////////// 685 686 // Flow attempts to move bytes from source to sink via buffer of size 687 // buffer_len. The function returns SR_SUCCESS when source reaches 688 // end-of-stream (returns SR_EOS), and all the data has been written successful 689 // to sink. Alternately, if source returns SR_BLOCK or SR_ERROR, or if sink 690 // returns SR_BLOCK, SR_ERROR, or SR_EOS, then the function immediately returns 691 // with the unexpected StreamResult value. 692 // data_len is the length of the valid data in buffer. in case of error 693 // this is the data that read from source but can't move to destination. 694 // as a pass in parameter, it indicates data in buffer that should move to sink 695 StreamResult Flow(StreamInterface* source, 696 char* buffer, size_t buffer_len, 697 StreamInterface* sink, size_t* data_len = NULL); 698 699 /////////////////////////////////////////////////////////////////////////////// 700 701 } // namespace rtc 702 703 #endif // WEBRTC_BASE_STREAM_H_ 704