message_queue.h 3.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091
  1. // Copyright 2016 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #ifndef MOJO_CORE_PORTS_MESSAGE_QUEUE_H_
  5. #define MOJO_CORE_PORTS_MESSAGE_QUEUE_H_
  6. #include <stdint.h>
  7. #include <limits>
  8. #include <memory>
  9. #include <vector>
  10. #include "base/component_export.h"
  11. #include "mojo/core/ports/event.h"
  12. namespace mojo {
  13. namespace core {
  14. namespace ports {
  15. constexpr uint64_t kInitialSequenceNum = 1;
  16. constexpr uint64_t kInvalidSequenceNum = std::numeric_limits<uint64_t>::max();
  17. class MessageFilter;
  18. // An incoming message queue for a port. MessageQueue keeps track of the highest
  19. // known sequence number and can indicate whether the next sequential message is
  20. // available. Thus the queue enforces message ordering for the consumer without
  21. // enforcing it for the producer (see AcceptMessage() below.)
  22. class COMPONENT_EXPORT(MOJO_CORE_PORTS) MessageQueue {
  23. public:
  24. explicit MessageQueue();
  25. explicit MessageQueue(uint64_t next_sequence_num);
  26. MessageQueue(const MessageQueue&) = delete;
  27. MessageQueue& operator=(const MessageQueue&) = delete;
  28. ~MessageQueue();
  29. void set_signalable(bool value) { signalable_ = value; }
  30. uint64_t next_sequence_num() const { return next_sequence_num_; }
  31. bool HasNextMessage() const;
  32. // Gives ownership of the message. If |filter| is non-null, the next message
  33. // will only be retrieved if the filter successfully matches it.
  34. // Need to call |MessageProcessed| after processing is finished.
  35. void GetNextMessage(std::unique_ptr<UserMessageEvent>* message,
  36. MessageFilter* filter);
  37. // Mark the message from |GetNextMessage| as processed.
  38. void MessageProcessed();
  39. // Takes ownership of the message. Note: Messages are ordered, so while we
  40. // have added a message to the queue, we may still be waiting on a message
  41. // ahead of this one before we can let any of the messages be returned by
  42. // GetNextMessage.
  43. //
  44. // Furthermore, once has_next_message is set to true, it will remain false
  45. // until GetNextMessage is called enough times to return a null message.
  46. // In other words, has_next_message acts like an edge trigger.
  47. //
  48. void AcceptMessage(std::unique_ptr<UserMessageEvent> message,
  49. bool* has_next_message);
  50. // Takes all messages from this queue. Used to safely destroy queued messages
  51. // without holding any Port lock.
  52. void TakeAllMessages(
  53. std::vector<std::unique_ptr<UserMessageEvent>>* messages);
  54. // The number of messages queued here, regardless of whether the next expected
  55. // message has arrived yet.
  56. size_t queued_message_count() const { return heap_.size(); }
  57. // The aggregate memory size in bytes of all messages queued here, regardless
  58. // of whether the next expected message has arrived yet.
  59. size_t queued_num_bytes() const { return total_queued_bytes_; }
  60. private:
  61. std::vector<std::unique_ptr<UserMessageEvent>> heap_;
  62. uint64_t next_sequence_num_;
  63. bool signalable_ = true;
  64. size_t total_queued_bytes_ = 0;
  65. };
  66. } // namespace ports
  67. } // namespace core
  68. } // namespace mojo
  69. #endif // MOJO_CORE_PORTS_MESSAGE_QUEUE_H_