diff options
Diffstat (limited to 'sensor/main')
-rw-r--r-- | sensor/main/io.cpp | 47 |
1 files changed, 26 insertions, 21 deletions
diff --git a/sensor/main/io.cpp b/sensor/main/io.cpp index cc713b9..e73b9d4 100644 --- a/sensor/main/io.cpp +++ b/sensor/main/io.cpp @@ -1,6 +1,7 @@ #include "trygvis/sensor/io.h" #include <map> +#include <chrono> #include "json.hpp" #include "boost/tokenizer.hpp" #include <boost/algorithm/string.hpp> @@ -10,6 +11,7 @@ namespace sensor { namespace io { using namespace std; +using namespace std::chrono; using boost::tokenizer; using boost::escaped_list_separator; using json = nlohmann::json; @@ -24,6 +26,22 @@ void ThreadSafeSampleOutputStream::write(SampleRecord const &sample) { underlying->write(sample); } + +AddTimestampSampleOutputStream::AddTimestampSampleOutputStream(unique_ptr<SampleOutputStream> underlying, + KeyDictionary &dict, + const string ×tamp_name) : underlying_(move(underlying)), timestamp_key(dict.indexOf(timestamp_name)) { +} + +void AddTimestampSampleOutputStream::write(SampleRecord const &sample) { + auto time_since_epoch = system_clock::now().time_since_epoch(); + auto timestamp = duration_cast<seconds>(time_since_epoch).count(); + auto timestamp_s = std::to_string(timestamp); + + SampleRecord copy = sample; + copy.set(timestamp_key, timestamp_s); + underlying_->write(copy); +} + void VectorSampleOutputStream::write(SampleRecord const &sample) { if (sample.empty()) { return; @@ -295,7 +313,7 @@ void SqlSampleOutputStream::write(SampleRecord const &values) { // (*stream.get()) << "INSERT INTO " << table_name << "(" << fs << ") VALUES(" << vs << ");" << endl << flush; } -void KeyValueSampleStreamParser::process(mutable_buffers_1 buffer) { +void KeyValueSampleStreamParser::process(mutable_buffers_1 &buffer) { size_t size = buffer_size(buffer); @@ -326,7 +344,7 @@ void KeyValueSampleStreamParser::process_line(shared_ptr<vector<uint8_t>> packet typedef tokenizer<escaped_list_separator<char>> Tokenizer; Tokenizer tokens(s); - SampleRecord sample(dict); + SampleRecord sample(dict_); for (auto token : tokens) { auto index = token.find('='); @@ -340,7 +358,7 @@ void KeyValueSampleStreamParser::process_line(shared_ptr<vector<uint8_t>> packet auto value = token.substr(index + 1); boost::algorithm::trim(value); - auto key = dict.indexOf(name); + auto key = dict_.indexOf(name); sample.set(key, value); } @@ -348,13 +366,13 @@ void KeyValueSampleStreamParser::process_line(shared_ptr<vector<uint8_t>> packet } AutoSampleParser::AutoSampleParser(shared_ptr<SampleOutputStream> output, KeyDictionary &dict) : - SampleStreamParser(sample_format_type::AUTO), keyValueParser(new KeyValueSampleStreamParser(output, dict)) { + SampleStreamParser(sample_format_type::AUTO, dict), keyValueParser(new KeyValueSampleStreamParser(output, dict)) { // Directly select the parser now until we have more than one parser parser = std::move(keyValueParser); type_ = sample_format_type::KEY_VALUE; } -void AutoSampleParser::process(mutable_buffers_1 buffer) { +void AutoSampleParser::process(mutable_buffers_1 &buffer) { if (parser) { parser->process(buffer); } else { @@ -375,24 +393,11 @@ unique_ptr<SampleStreamParser> open_sample_stream_parser( } } -template<typename T> -o<T *> find_option(vector<sample_output_stream_option *> &options) { - for (sample_output_stream_option *&option : options) { - T *x = dynamic_cast<T *>(option); - - if (x != nullptr) { - return o<T *>(x); - } - } - - return o<T *>(); -} - unique_ptr<SampleOutputStream> open_sample_output_stream( shared_ptr<ostream> output, KeyDictionary &dict, sample_format_type type, - vector<sample_output_stream_option *> options) { + sample_output_stream_options options) { if (type == sample_format_type::CSV) { return make_unique<CsvSampleOutputStream>(output, dict); @@ -401,9 +406,9 @@ unique_ptr<SampleOutputStream> open_sample_output_stream( } else if (type == sample_format_type::JSON) { return make_unique<JsonSampleOutputStream>(output, dict); } else if (type == sample_format_type::RRD) { - o<output_fields_option *> of = find_option<output_fields_option>(options); + o<output_fields_option *> of = options.find_option<output_fields_option>(); - o<timestamp_field_option *> tsf = find_option<timestamp_field_option>(options); + o<timestamp_field_option *> tsf = options.find_option<timestamp_field_option>(); auto timestamp_key = dict.indexOf(tsf ? tsf.get()->name : "timestamp"); |