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

#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;
}