wasm_message_queue.h

Go to the documentation of this file.

#ifndef YAZE_APP_PLATFORM_WASM_MESSAGE_QUEUE_H_
#define YAZE_APP_PLATFORM_WASM_MESSAGE_QUEUE_H_
 
#ifdef __EMSCRIPTEN__
 
#include <emscripten.h>
#include <emscripten/val.h>
 
#include <chrono>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <vector>
 
#include "absl/status/status.h"
#include "nlohmann/json.hpp"
 
namespace yaze {
namespace app {
namespace platform {
 
class WasmMessageQueue {
 public:
  struct QueuedMessage {
    std::string message_type;  // "change", "cursor", etc.
    std::string payload;        // JSON payload
    double timestamp;           // When message was queued
    int retry_count = 0;        // Number of send attempts
    std::string id;             // Unique message ID
  };
 
  struct QueueStatus {
    size_t pending_count = 0;
    size_t failed_count = 0;
    size_t total_bytes = 0;
    double oldest_message_age = 0;  // Seconds
    bool is_persisted = false;
  };
 
  // Callback types
  using ReplayCompleteCallback = std::function<void(int replayed_count, int failed_count)>;
  using MessageSender = std::function<absl::Status(const std::string& type, const std::string& payload)>;
  using StatusChangeCallback = std::function<void(const QueueStatus& status)>;
 
  WasmMessageQueue();
  ~WasmMessageQueue();
 
  std::string Enqueue(const std::string& message_type, const std::string& payload);
 
  void SetOnReplayComplete(ReplayCompleteCallback callback) {
    replay_complete_callback_ = callback;
  }
 
  void SetOnStatusChange(StatusChangeCallback callback) {
    status_change_callback_ = callback;
  }
 
  void ReplayAll(MessageSender sender, int max_retries = 3);
 
  size_t PendingCount() const;
 
  QueueStatus GetStatus() const;
 
  void Clear();
 
  void ClearFailed();
 
  bool RemoveMessage(const std::string& message_id);
 
  absl::Status PersistToStorage();
 
  absl::Status LoadFromStorage();
 
  void SetAutoPersist(bool enable) {
    auto_persist_ = enable;
  }
 
  void SetMaxQueueSize(size_t max_size) {
    max_queue_size_ = max_size;
  }
 
  void SetMessageExpiry(double seconds) {
    message_expiry_seconds_ = seconds;
  }
 
  std::vector<QueuedMessage> GetQueuedMessages() const;
 
  int PruneExpiredMessages();
 
 private:
  // Generate unique message ID
  std::string GenerateMessageId();
 
  // Calculate total size of queued messages
  size_t CalculateTotalBytes() const;
 
  // Notify status change listeners
  void NotifyStatusChange();
 
  // Check if we should persist to storage
  void MaybePersist();
 
  // Message queue
  std::deque<QueuedMessage> queue_;
  std::vector<QueuedMessage> failed_messages_;
  mutable std::mutex queue_mutex_;
 
  // Configuration
  bool auto_persist_ = true;
  size_t max_queue_size_ = 1000;
  double message_expiry_seconds_ = 86400.0;  // 24 hours
 
  // State tracking
  bool is_replaying_ = false;
  size_t total_enqueued_ = 0;
  size_t total_replayed_ = 0;
  size_t total_failed_ = 0;
 
  // Callbacks
  ReplayCompleteCallback replay_complete_callback_;
  StatusChangeCallback status_change_callback_;
 
  // Storage key for IndexedDB
  static constexpr const char* kStorageKey = "collaboration_message_queue";
};
 
}  // namespace platform
}  // namespace app
}  // namespace yaze
 
#else  // !__EMSCRIPTEN__
 
// Stub implementation for non-WASM builds
#include <deque>
#include <functional>
#include <string>
#include <vector>
 
#include "absl/status/status.h"
 
namespace yaze {
namespace app {
namespace platform {
 
class WasmMessageQueue {
 public:
  struct QueuedMessage {
    std::string message_type;
    std::string payload;
    double timestamp;
    int retry_count = 0;
    std::string id;
  };
 
  struct QueueStatus {
    size_t pending_count = 0;
    size_t failed_count = 0;
    size_t total_bytes = 0;
    double oldest_message_age = 0;
    bool is_persisted = false;
  };
 
  using ReplayCompleteCallback = std::function<void(int, int)>;
  using MessageSender = std::function<absl::Status(const std::string&, const std::string&)>;
  using StatusChangeCallback = std::function<void(const QueueStatus&)>;
 
  WasmMessageQueue() {}
  ~WasmMessageQueue() {}
 
  std::string Enqueue(const std::string&, const std::string&) { return ""; }
  void SetOnReplayComplete(ReplayCompleteCallback) {}
  void SetOnStatusChange(StatusChangeCallback) {}
  void ReplayAll(MessageSender, int = 3) {}
  size_t PendingCount() const { return 0; }
  QueueStatus GetStatus() const { return {}; }
  void Clear() {}
  void ClearFailed() {}
  bool RemoveMessage(const std::string&) { return false; }
  absl::Status PersistToStorage() {
    return absl::UnimplementedError("Message queue requires WASM build");
  }
  absl::Status LoadFromStorage() {
    return absl::UnimplementedError("Message queue requires WASM build");
  }
  void SetAutoPersist(bool) {}
  void SetMaxQueueSize(size_t) {}
  void SetMessageExpiry(double) {}
  std::vector<QueuedMessage> GetQueuedMessages() const { return {}; }
  int PruneExpiredMessages() { return 0; }
};
 
}  // namespace platform
}  // namespace app
}  // namespace yaze
 
#endif  // __EMSCRIPTEN__
 
#endif  // YAZE_APP_PLATFORM_WASM_MESSAGE_QUEUE_H_