LiteLib/tests/Memcached/src/lite-version/include/service.hpp at master · OrderLab/LiteLib · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
#pragma once

#include <event.h>

#include <lite.hpp>
#include <memory>
#include <string>

#include "packet.hpp"

struct CacheEntry {
  std::shared_ptr<std::vector<uint8_t>> value = nullptr;
  std::shared_ptr<std::vector<uint8_t>> flags = nullptr;
  uint64_t CAS;
  size_t GetSize() const {
    return (value ? value->size() : 0) + (flags ? flags->size() : 0) +
           sizeof(CAS);
  }

  std::shared_ptr<Packet> ToRequest(const std::vector<uint8_t> &key) const {
    auto req = std::make_shared<ParsedPacket>();
    static std::vector<uint8_t> expiry(4, 0);  // TODO: use real one
    req->header.magic = 0x80;
    req->header.opcode = magic_enum::enum_underlying(Header::Opcode::kSet);
    req->key = std::make_shared<std::vector<uint8_t>>(key);
    req->value = value;
    req->extra = flags;
    req->header.CAS = CAS;
    req->header.extras_length = 8;
    req->header.key_length = req->key->size();
    req->header.total_body_length =
        req->value->size() + req->header.key_length + req->header.extras_length;
    req->buffer->clear();
    return req;
  }
};

struct LogEntry {  // TODO: deal with expiry
  std::shared_ptr<Packet> packet;

  std::shared_ptr<std::vector<uint8_t>> Serialize() {
    return packet->Serialize();
  }

  lite::DeserializeResult Deserialize(uint8_t *&begin, uint8_t *end) {
    return packet->Deserialize(begin, end);
  }
};

struct ConnectionInfo {
  std::unique_ptr<std::vector<uint8_t>> response_buffer;
  ConnectionInfo()
      : response_buffer(std::make_unique<std::vector<uint8_t>>()) {}
};

class Memcached {
  using Cache = lite::Cache<Memcached, Packet, Packet, ConnectionInfo,
                            std::vector<uint8_t>, CacheEntry>;
  using Logger = lite::Logger<Memcached, Packet, Packet, ConnectionInfo,
                              std::vector<uint8_t>, CacheEntry>;

 public:
  std::pair<std::vector<std::shared_ptr<Packet>>, bool> Match(
      const std::shared_ptr<Packet> &resp, ConnectionInfo &_,
      lite::ThreadSafeQueue<std::pair<std::shared_ptr<Packet>, bool>>
          &pending_requests) const;

  void NormalUpdate(const std::shared_ptr<Packet> &resp,
                    std::vector<std::shared_ptr<Packet>> requests,
                    ConnectionInfo &_, Cache *cache) const;

  void HandleReplayResponse(const std::shared_ptr<Packet> &resp,
                            std::vector<std::shared_ptr<Packet>> requests,
                            ConnectionInfo &_, Cache *cache) const;

  std::pair<Packet, bool> EmergencyServe(std::shared_ptr<Packet> p,
                                         ConnectionInfo &conn_info,
                                         Cache *cache, Logger *logger,
                                         bool flow_control) const;

  void NormalToEmergencyHook() {}

  void EmergencyToNormalHook() {}

  Packet EmergencyConnectionEstablishHook(ConnectionInfo &conn_info) {
    Packet packet = Packet();
    packet.buffer = nullptr;
    return packet;
  }
};