MercurySession.h 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128
  1. #pragma once
  2. #include <atomic>
  3. #include <functional>
  4. #include <memory>
  5. #include <string>
  6. #include <unordered_map>
  7. #include <vector>
  8. #include "BellTask.h"
  9. #include "Logger.h"
  10. #include "NanoPBHelper.h"
  11. #include "Packet.h"
  12. #include "Queue.h"
  13. #include "Session.h"
  14. #include "TimeProvider.h"
  15. #include "Utils.h"
  16. #include "protobuf/mercury.pb.h"
  17. namespace cspot {
  18. class MercurySession : public bell::Task, public cspot::Session {
  19. public:
  20. MercurySession(std::shared_ptr<cspot::TimeProvider> timeProvider);
  21. ~MercurySession();
  22. typedef std::vector<std::vector<uint8_t>> DataParts;
  23. struct Response {
  24. Header mercuryHeader;
  25. uint8_t flags;
  26. DataParts parts;
  27. uint64_t sequenceId;
  28. bool fail;
  29. };
  30. typedef std::function<void(Response&)> ResponseCallback;
  31. typedef std::function<void(bool, const std::vector<uint8_t>&)> AudioKeyCallback;
  32. typedef std::function<void()> ConnectionEstabilishedCallback;
  33. enum class RequestType : uint8_t {
  34. SUB = 0xb3,
  35. UNSUB = 0xb4,
  36. SUBRES = 0xb5,
  37. SEND = 0xb2,
  38. GET = 0xFF, // Shitty workaround, it's value is actually same as SEND
  39. PING = 0x04,
  40. PONG_ACK = 0x4a,
  41. AUDIO_CHUNK_REQUEST_COMMAND = 0x08,
  42. AUDIO_CHUNK_SUCCESS_RESPONSE = 0x09,
  43. AUDIO_CHUNK_FAILURE_RESPONSE = 0x0A,
  44. AUDIO_KEY_REQUEST_COMMAND = 0x0C,
  45. AUDIO_KEY_SUCCESS_RESPONSE = 0x0D,
  46. AUDIO_KEY_FAILURE_RESPONSE = 0x0E,
  47. COUNTRY_CODE_RESPONSE = 0x1B,
  48. };
  49. std::unordered_map<RequestType, std::string> RequestTypeMap = {
  50. {RequestType::GET, "GET"},
  51. {RequestType::SEND, "SEND"},
  52. {RequestType::SUB, "SUB"},
  53. {RequestType::UNSUB, "UNSUB"},
  54. };
  55. void handlePacket();
  56. uint64_t executeSubscription(RequestType type, const std::string& uri,
  57. ResponseCallback callback,
  58. ResponseCallback subscription, DataParts& parts);
  59. uint64_t executeSubscription(RequestType type, const std::string& uri,
  60. ResponseCallback callback,
  61. ResponseCallback subscription) {
  62. DataParts parts = {};
  63. return this->executeSubscription(type, uri, callback, subscription, parts);
  64. }
  65. uint64_t execute(RequestType type, const std::string& uri,
  66. ResponseCallback callback) {
  67. return this->executeSubscription(type, uri, callback, nullptr);
  68. }
  69. uint64_t execute(RequestType type, const std::string& uri,
  70. ResponseCallback callback, DataParts& parts) {
  71. return this->executeSubscription(type, uri, callback, nullptr, parts);
  72. }
  73. void requestAudioKey(const std::vector<uint8_t>& trackId,
  74. const std::vector<uint8_t>& fileId,
  75. AudioKeyCallback audioCallback);
  76. std::string getCountryCode();
  77. void disconnect();
  78. void setConnectedHandler(ConnectionEstabilishedCallback callback);
  79. bool triggerTimeout() override;
  80. private:
  81. const int PING_TIMEOUT_MS = 2 * 60 * 1000 + 5000;
  82. std::shared_ptr<cspot::TimeProvider> timeProvider;
  83. Header tempMercuryHeader = {};
  84. ConnectionEstabilishedCallback connectionReadyCallback = nullptr;
  85. bell::Queue<cspot::Packet> packetQueue;
  86. void runTask() override;
  87. void reconnect();
  88. std::unordered_map<uint64_t, ResponseCallback> callbacks;
  89. std::unordered_map<std::string, ResponseCallback> subscriptions;
  90. AudioKeyCallback audioKeyCallback;
  91. uint64_t sequenceId = 1;
  92. uint32_t audioKeySequence = 1;
  93. unsigned long long timestampDiff;
  94. unsigned long long lastPingTimestamp = -1;
  95. std::string countryCode = "";
  96. std::mutex isRunningMutex;
  97. std::atomic<bool> isRunning = false;
  98. std::atomic<bool> isReconnecting = false;
  99. std::atomic<bool> executeEstabilishedCallback = false;
  100. void failAllPending();
  101. Response decodeResponse(const std::vector<uint8_t>& data);
  102. };
  103. } // namespace cspot