redisadapter/RedisAdapter.hpp at main · bigsamich/redisadapter · 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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
/**
 * RedisAdapter.H
 *
 * This file contains the definition of the redis Adapter class.
 *
 * @author rsantucc
 */

#ifndef RedisAdapter_HPP
#define RedisAdapter_HPP

#include <string>
#include <errno.h>
#include <stdexcept>
#include <unistd.h>
#include <sstream>
#include <iostream>
#include <thread>
#include <sw/redis++/redis++.h>
#include "IRedisAdapter.hpp"
#include <TRACE/trace.h>

using namespace std;
using namespace sw::redis;
  /**
   * RedisAdapter
   */
class RedisAdapter: public IRedisAdapter {

   public:
	/*Constructor / Destructor*/
    RedisAdapter(string key, string connection = "tcp://127.0.0.1:6380");
	RedisAdapter(const RedisAdapter& ra);
    ~RedisAdapter();

	/*Wrapper Functions*/
	virtual vector<string> getDevices();
    virtual void clearDevices(string devicelist);
	virtual void setDeviceConfig(std::unordered_map<std::string, std::string> map);
	virtual std::unordered_map<std::string, std::string> getDeviceConfig();
	virtual void setDevice(string name);

	/* Single Value Functions
	*  Note: These use the config connection
	*
	*/
	virtual string getValue(string key);
	virtual void setValue(string key, string val);
	virtual int getUniqueValue(string key);
	virtual unordered_map<string, string> getHash(string key);
	virtual void setHash(string key, std::unordered_map<std::string, std::string> m);
	virtual std::unordered_set<string> getSet(string key);
	virtual void setSet(string key, string val);

	/*
	* Stream Functions
	* Note: All stream functions use the cluster connection.
	*		logRead and logWrite are stream functions, but use the config connection
	*/

	virtual void streamWrite(vector<pair<string,string>> data, string timeID, string key, uint trim = 0);
	virtual string streamReadBlock(std::unordered_map<string,string> keysID, int count, std::unordered_map<string,vector<float>>& result);
	virtual void streamRead(string key, string time, int count, vector<float>& result);
	virtual void streamRead(string key, string time, int count, ItemStream& dest);
	virtual void streamTrim(string key, int size);
	virtual ItemStream logRead(uint count);
	virtual void logWrite(string key, string msg, string source);

	/*
	* Publish / Subscribe Functions
	* Note: All publish / subscribe functions use the config connection
	*/
	virtual void publish(string msg);
	virtual void publish(string key, string msg);
	virtual void psubscribe(std::string pattern, std::function<void(std::string,std::string,std::string)> f);
	virtual void subscribe(std::string channel, std::function<void(std::string,std::string)> f);
	virtual void registerCommand(std::string command, std::function<void(std::string, std::string)> f);

	/*
	* Copy Functions
	*/
	virtual void copyKey(string src, string dest, bool data = false);

	/*
	*	Abort Flag
	*/
	virtual void setAbortFlag(bool flag = false);
	virtual bool getAbortFlag();

	/*
	* Time
	*/
	virtual vector<string> getServerTime();

	/* Key getters and setters*/
	virtual string getBaseKey() const { return _baseKey;}
	virtual void setBaseKey(string baseKey) { _baseKey = baseKey;}

	virtual string getChannelKey() const { return _channelKey;}
	virtual void setChannelKey(string channelKey) { _channelKey = channelKey;}

	virtual	string getConfigKey()  const {return _configKey;}
	virtual void setConfigKey(string configKey) { _configKey = configKey;}

	virtual string getLogKey()  const   {return _logKey;}
	virtual void setLogKey(string logKey) { _logKey = logKey;}

	virtual string getStatusKey() const {return _statusKey;}
	virtual void setStatusKey(string statusKey) { _statusKey = statusKey;}

	virtual string getTimeKey()  const  {return _timeKey;}
	virtual void setTimeKey(string timeKey) { _timeKey = timeKey;}

	virtual string getDeviceKey()  const  {return _deviceKey;}
	virtual void setDeviceKey(string deviceKey) { _deviceKey = deviceKey;}

	virtual string getDataBaseKey() const {return _dataBaseKey;}
	virtual void setDataBaseKey(string dataBaseKey) {_dataBaseKey = dataBaseKey;}
	virtual string getDataKey(string subkey) {return _dataBaseKey + ":" + subkey;}

	virtual string getAbortKey()  const  {return _abortKey;}
	virtual void setAbortKey(string abortKey) { _abortKey = abortKey;}


	void	listener();
	void	reader();
	virtual void startListener();
	virtual void startReader();

	virtual bool getDeviceStatus();
	virtual void setDeviceStatus(bool status = true);

    RedisCluster _redisCluster;

	thread _reader;
    thread _listener;


	map<string, std::function<void(std::string,std::string,std::string)>> patternSubscriptions;
	map<string, std::function<void(std::string,std::string)>> subscriptions;
	map<string, std::function<void(std::string,std::string)>> commands;


	vector<string> streamKeys;
	ItemStream buffer;
	std::mutex m_buffer;
	std::string _lastTimeID = "$";
    std::mutex m_lastTimeID;

	std::string  _baseKey, _configKey, _logKey, _channelKey, _statusKey, _timeKey, _deviceKey, _abortKey;
	std::string	 _dataBaseKey;
	std::string _connection;
  };

#endif