Closes #11

analysis/05-Message_Queue_Partitioning/example/message_queue_partitioning/main.cpp

@@ -0,0 +1,165 @@
+#include <cassert>
+#include <cstdint>
+#include <iostream>
+
+struct Message {
+  std::uint32_t id;
+  bool retry;
+  Message *next;
+};
+
+struct MessageQueue {
+  Message *head;
+  Message *tail;
+};
+
+struct PartitionResult {
+  MessageQueue ready;
+  MessageQueue retry;
+};
+
+static void append (MessageQueue &queue, Message *message) {
+  assert (message != nullptr);
+  assert (message->next == nullptr);
+
+  if (queue.tail == nullptr) {
+    queue.head = message;
+    queue.tail = message;
+    return;
+  }
+
+  queue.tail->next = message;
+  queue.tail = message;
+}
+
+PartitionResult partition_messages (MessageQueue &source) {
+  PartitionResult result{
+    {nullptr, nullptr},
+    {nullptr, nullptr}
+  };
+
+  Message *current = source.head;
+
+  /*
+   * The source queue is consumed by this operation.
+   *
+   * Clearing it before traversal makes the ownership transfer explicit:
+   * every node taken from the original queue must be appended to exactly
+   * one of the two result queues.
+   */
+  source.head = nullptr;
+  source.tail = nullptr;
+
+  while (current != nullptr) {
+    /*
+     * Save the traversal link before modifying current->next.
+     * The same intrusive link is reused by the destination queue.
+     */
+    Message *next = current->next;
+    current->next = nullptr;
+
+    if (current->retry)
+      append (result.retry, current);
+    else
+      append (result.ready, current);
+
+    current = next;
+  }
+
+  return result;
+}
+
+static void print_queue (const char *name, const MessageQueue &queue) {
+  std::cout << name << ": ";
+
+  const Message *current = queue.head;
+
+  if (current == nullptr) {
+    std::cout << "<empty>\n";
+    return;
+  }
+
+  while (current != nullptr) {
+    std::cout << current->id;
+
+    if (current->next != nullptr)
+      std::cout << " -> ";
+
+    current = current->next;
+  }
+
+  std::cout << '\n';
+}
+
+static std::size_t queue_size (const MessageQueue &queue) {
+  std::size_t size = 0;
+  const Message *current = queue.head;
+
+  while (current != nullptr) {
+    ++size;
+    current = current->next;
+  }
+
+  return size;
+}
+
+static void verify_queue (const MessageQueue &queue) {
+  if (queue.head == nullptr) {
+    assert (queue.tail == nullptr);
+    return;
+  }
+
+  assert (queue.tail != nullptr);
+  assert (queue.tail->next == nullptr);
+
+  const Message *current = queue.head;
+
+  while (current->next != nullptr)
+    current = current->next;
+
+  assert (current == queue.tail);
+}
+
+int main() {
+  Message a{1U, false, nullptr};
+  Message b{2U, true,  nullptr};
+  Message c{3U, false, nullptr};
+  Message d{4U, true,  nullptr};
+
+  a.next = &b;
+  b.next = &c;
+  c.next = &d;
+
+  MessageQueue outgoing{&a, &d};
+
+  std::cout << "Before partition\n";
+  print_queue ("Outgoing", outgoing);
+
+  const PartitionResult result = partition_messages (outgoing);
+
+  std::cout << "\nAfter partition\n";
+  print_queue ("Outgoing", outgoing);
+  print_queue ("Ready", result.ready);
+  print_queue ("Retry", result.retry);
+
+  verify_queue (outgoing);
+  verify_queue (result.ready);
+  verify_queue (result.retry);
+
+  assert (outgoing.head == nullptr);
+  assert (outgoing.tail == nullptr);
+
+  assert (result.ready.head == &a);
+  assert (result.ready.tail == &c);
+  assert (a.next == &c);
+  assert (c.next == nullptr);
+
+  assert (result.retry.head == &b);
+  assert (result.retry.tail == &d);
+  assert (b.next == &d);
+  assert (d.next == nullptr);
+
+  assert (queue_size (result.ready) + queue_size (result.retry) == 4U);
+
+  return 0;
+}