Skip to content

Commit 6ced9b0

Browse files
threecgreenthreecgreen
authored
VER: Release 0.18.1
2 parents ce8eb73 + 246e14d commit 6ced9b0

File tree

16 files changed

+251
-42
lines changed

16 files changed

+251
-42
lines changed

‎CHANGELOG.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,18 @@
11
# Changelog
22

3+
## 0.18.1 - 2024-05-22
4+
5+
### Enhancements
6+
- Added live `Subscribe` function overload with `use_snapshot` parameter
7+
- Added `GetIf` method to `Record` that allows `if` chaining for handling multiple
8+
record types
9+
- Added record type checking to `Record::Get` method to catch programming errors
10+
and prevent reading invalid data
11+
12+
### Bug fixes
13+
- Added missing symbol chunking for live `Subscribe` overloads with `const std::string&`
14+
`start` parameter
15+
316
## 0.18.0 - 2024-05-14
417

518
### Breaking changes

‎CMakeLists.txt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ cmake_minimum_required(VERSION 3.14)
44
# Project details
55
#
66

7-
project("databento" VERSION 0.18.0 LANGUAGES CXX)
7+
project("databento" VERSION 0.18.1 LANGUAGES CXX)
88
string(TOUPPER ${PROJECT_NAME} PROJECT_NAME_UPPERCASE)
99

1010
#

‎README.md‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -108,10 +108,9 @@ int main() {
108108

109109
auto handler = [&symbol_mappings](const Record& rec) {
110110
symbol_mappings.OnRecord(rec);
111-
if (rec.Holds<TradeMsg>()) {
112-
auto trade = rec.Get<TradeMsg>();
111+
if (const auto* trade = rec.GetIf<TradeMsg>()) {
113112
std::cout << "Received trade for "
114-
<< symbol_mappings[trade.hd.instrument_id] << ':' << trade
113+
<< symbol_mappings[trade->hd.instrument_id] << ':' << *trade
115114
<< '\n';
116115
}
117116
return KeepGoing::Continue;

‎example/live/readme.cpp‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,9 @@ int main() {
1818

1919
auto handler = [&symbol_mappings](const Record& rec) {
2020
symbol_mappings.OnRecord(rec);
21-
if (rec.Holds<TradeMsg>()) {
22-
auto trade = rec.Get<TradeMsg>();
21+
if (const auto* trade = rec.GetIf<TradeMsg>()) {
2322
std::cout << "Received trade for "
24-
<< symbol_mappings[trade.hd.instrument_id] << ':' << trade
23+
<< symbol_mappings[trade->hd.instrument_id] << ':' << *trade
2524
<< '\n';
2625
}
2726
return KeepGoing::Continue;

‎include/databento/live_blocking.hpp‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,8 @@ class LiveBlocking {
5050
SType stype_in, UnixNanos start);
5151
void Subscribe(const std::vector<std::string>& symbols, Schema schema,
5252
SType stype_in, const std::string& start);
53+
void Subscribe(const std::vector<std::string>& symbols, Schema schema,
54+
SType stype_in, bool use_snapshot);
5355
// Notifies the gateway to start sending messages for all subscriptions.
5456
//
5557
// This method should only be called once per instance.
@@ -78,6 +80,8 @@ class LiveBlocking {
7880
std::string GenerateCramReply(const std::string& challenge_key);
7981
std::string EncodeAuthReq(const std::string& auth);
8082
std::uint64_t DecodeAuthResp();
83+
void Subscribe(const std::string& sub_msg,
84+
const std::vector<std::string>& symbols, bool use_snapshot);
8185
detail::TcpClient::Result FillBuffer(std::chrono::milliseconds timeout);
8286
RecordHeader* BufferRecordHeader();
8387

‎include/databento/live_threaded.hpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,8 @@ class LiveThreaded {
6565
SType stype_in, UnixNanos start);
6666
void Subscribe(const std::vector<std::string>& symbols, Schema schema,
6767
SType stype_in, const std::string& start);
68+
void Subscribe(const std::vector<std::string>& symbols, Schema schema,
69+
SType stype_in, bool use_snapshot);
6870
// Notifies the gateway to start sending messages for all subscriptions.
6971
// `metadata_callback` will be called exactly once, before any calls to
7072
// `record_callback`. `record_callback` will be called for records from all

‎include/databento/record.hpp‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
#include "databento/constants.hpp" // kSymbolCstrLen
1111
#include "databento/datetime.hpp" // UnixNanos
1212
#include "databento/enums.hpp"
13+
#include "databento/exceptions.hpp" // InvalidArgumentError
1314
#include "databento/flag_set.hpp" // FlagSet
1415
#include "databento/publishers.hpp" // Publisher
1516
#include "databento/with_ts_out.hpp"
@@ -54,11 +55,33 @@ class Record {
5455

5556
template <typename T>
5657
const T& Get() const {
57-
return *reinterpret_cast<const T*>(record_);
58+
if (const auto* r = GetIf<T>()) {
59+
return *r;
60+
}
61+
throw InvalidArgumentError{
62+
"Get", "T", std::string{"rtype mismatch, found "} + ToString(RType())};
5863
}
5964
template <typename T>
6065
T& Get() {
61-
return *reinterpret_cast<T*>(record_);
66+
if (auto* r = GetIf<T>()) {
67+
return *r;
68+
}
69+
throw InvalidArgumentError{
70+
"Get", "T", std::string{"rtype mismatch, found "} + ToString(RType())};
71+
}
72+
template <typename T>
73+
const T* GetIf() const {
74+
if (!Holds<T>()) {
75+
return nullptr;
76+
}
77+
return reinterpret_cast<const T*>(record_);
78+
}
79+
template <typename T>
80+
T* GetIf() {
81+
if (!Holds<T>()) {
82+
return nullptr;
83+
}
84+
return reinterpret_cast<T*>(record_);
6285
}
6386

6487
std::size_t Size() const;
@@ -629,6 +652,8 @@ inline bool operator!=(const SymbolMappingMsg& lhs,
629652

630653
std::string ToString(const RecordHeader& header);
631654
std::ostream& operator<<(std::ostream& stream, const RecordHeader& header);
655+
std::string ToString(const Record& header);
656+
std::ostream& operator<<(std::ostream& stream, const Record& header);
632657
std::string ToString(const MboMsg& mbo_msg);
633658
std::ostream& operator<<(std::ostream& stream, const MboMsg& mbo_msg);
634659
std::string ToString(const BidAskPair& ba_pair);

‎pkg/PKGBUILD‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
# Maintainer: Databento <support@databento.com>
22
_pkgname=databento-cpp
33
pkgname=databento-cpp-git
4-
pkgver=0.18.0
4+
pkgver=0.18.1
55
pkgrel=1
66
pkgdesc="Official C++ client for Databento"
77
arch=('any')

‎src/live_blocking.cpp‎

Lines changed: 40 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,42 @@ void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
5959

6060
void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
6161
Schema schema, SType stype_in, UnixNanos start) {
62+
std::ostringstream sub_msg;
63+
sub_msg << "schema=" << ToString(schema)
64+
<< "|stype_in=" << ToString(stype_in);
65+
if (start.time_since_epoch().count()) {
66+
sub_msg << "|start=" << start.time_since_epoch().count();
67+
}
68+
Subscribe(sub_msg.str(), symbols, false);
69+
}
70+
71+
void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
72+
Schema schema, SType stype_in,
73+
const std::string& start) {
74+
std::ostringstream sub_msg;
75+
sub_msg << "schema=" << ToString(schema)
76+
<< "|stype_in=" << ToString(stype_in);
77+
if (!start.empty()) {
78+
sub_msg << "|start=" << start;
79+
}
80+
Subscribe(sub_msg.str(), symbols, false);
81+
}
82+
83+
void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
84+
Schema schema, SType stype_in, bool use_snapshot) {
85+
std::ostringstream sub_msg;
86+
sub_msg << "schema=" << ToString(schema)
87+
<< "|stype_in=" << ToString(stype_in);
88+
89+
Subscribe(sub_msg.str(), symbols, use_snapshot);
90+
}
91+
92+
void LiveBlocking::Subscribe(const std::string& sub_msg,
93+
const std::vector<std::string>& symbols,
94+
bool use_snapshot) {
6295
static constexpr auto kMethodName = "Live::Subscribe";
6396
constexpr std::ptrdiff_t kSymbolMaxChunkSize = 128;
97+
6498
if (symbols.empty()) {
6599
throw InvalidArgumentError{kMethodName, "symbols",
66100
"must contain at least one symbol"};
@@ -70,36 +104,17 @@ void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
70104
const auto chunk_size =
71105
std::min(kSymbolMaxChunkSize, std::distance(symbols_it, symbols.end()));
72106

73-
std::ostringstream sub_msg;
74-
sub_msg << "schema=" << ToString(schema)
75-
<< "|stype_in=" << ToString(stype_in) << "|symbols="
76-
<< JoinSymbolStrings(kMethodName, symbols_it,
77-
symbols_it + chunk_size);
78-
if (start.time_since_epoch().count()) {
79-
sub_msg << "|start=" << start.time_since_epoch().count();
80-
}
81-
sub_msg << '\n';
82-
client_.WriteAll(sub_msg.str());
107+
std::ostringstream chunked_sub_msg;
108+
chunked_sub_msg << sub_msg << "|symbols="
109+
<< JoinSymbolStrings(kMethodName, symbols_it,
110+
symbols_it + chunk_size)
111+
<< "|snapshot=" << use_snapshot << '\n';
112+
client_.WriteAll(chunked_sub_msg.str());
83113

84114
symbols_it += chunk_size;
85115
}
86116
}
87117

88-
void LiveBlocking::Subscribe(const std::vector<std::string>& symbols,
89-
Schema schema, SType stype_in,
90-
const std::string& start) {
91-
std::ostringstream sub_msg;
92-
sub_msg << "schema=" << ToString(schema) << "|stype_in=" << ToString(stype_in)
93-
<< "|symbols="
94-
<< JoinSymbolStrings("LiveBlocking::Subscribe", symbols);
95-
if (!start.empty()) {
96-
sub_msg << "|start=" << start;
97-
}
98-
sub_msg << '\n';
99-
100-
client_.WriteAll(sub_msg.str());
101-
}
102-
103118
databento::Metadata LiveBlocking::Start() {
104119
constexpr auto kMetadataPreludeSize = 8;
105120
client_.WriteAll("start_session\n");

‎src/live_threaded.cpp‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,11 @@ void LiveThreaded::Subscribe(const std::vector<std::string>& symbols,
102102
impl_->blocking.Subscribe(symbols, schema, stype_in, start);
103103
}
104104

105+
void LiveThreaded::Subscribe(const std::vector<std::string>& symbols,
106+
Schema schema, SType stype_in, bool use_snapshot) {
107+
impl_->blocking.Subscribe(symbols, schema, stype_in, use_snapshot);
108+
}
109+
105110
void LiveThreaded::Start(RecordCallback callback) {
106111
Start({}, std::move(callback), {});
107112
}

0 commit comments

Comments
 (0)