| 620 | } |
| 621 | |
| 622 | void _processLogMessages() { |
| 623 | std::vector<LogsDBRequest> requests; |
| 624 | std::vector<LogsDBResponse> responses; |
| 625 | auto& requestMessages = _channel.protocolMessages(LOG_REQ_PROTOCOL_VERSION); |
| 626 | auto& responseMessages = _channel.protocolMessages(LOG_RESP_PROTOCOL_VERSION); |
| 627 | requests.reserve(requestMessages.size()); |
| 628 | responses.reserve(responseMessages.size()); |
| 629 | for (auto& msg : requestMessages) { |
| 630 | auto replicaId = _getReplicaId(msg.clientAddr); |
| 631 | if (replicaId == LogsDB::REPLICA_COUNT) { |
| 632 | LOG_DEBUG(_env, "We can't match this address (%s) to replica. Dropping", msg.clientAddr); |
| 633 | continue; |
| 634 | } |
| 635 | auto& req = requests.emplace_back(); |
| 636 | req.replicaId = replicaId; |
| 637 | try { |
| 638 | req.msg.unpack(msg.buf, _expandedCDCKey); |
| 639 | } catch (const BincodeException& err) { |
| 640 | LOG_ERROR(_env, "could not parse: %s", err.what()); |
| 641 | RAISE_ALERT(_env, "could not parse LogsDBRequest from %s, dropping it.", msg.clientAddr); |
| 642 | requests.pop_back(); |
| 643 | continue; |
| 644 | } |
| 645 | LOG_DEBUG(_env, "Received request %s with requests id %s from replica id %s", req.msg.body.kind(), req.msg.id, req.replicaId); |
| 646 | } |
| 647 | for (auto& msg : responseMessages) { |
| 648 | auto replicaId = _getReplicaId(msg.clientAddr); |
| 649 | if (replicaId == LogsDB::REPLICA_COUNT) { |
| 650 | LOG_DEBUG(_env, "We can't match this address (%s) to replica. Dropping", msg.clientAddr); |
| 651 | continue; |
| 652 | } |
| 653 | auto& resp = responses.emplace_back(); |
| 654 | resp.replicaId = replicaId; |
| 655 | try { |
| 656 | resp.msg.unpack(msg.buf, _expandedCDCKey); |
| 657 | } catch (const BincodeException& err) { |
| 658 | LOG_ERROR(_env, "could not parse: %s", err.what()); |
| 659 | RAISE_ALERT(_env, "could not parse LogsDBResponse from %s, dropping it.", msg.clientAddr); |
| 660 | requests.pop_back(); |
| 661 | continue; |
| 662 | } |
| 663 | LOG_DEBUG(_env, "Received response %s with requests id %s from replica id %s", resp.msg.body.kind(), resp.msg.id, resp.replicaId); |
| 664 | } |
| 665 | _logsDB.processIncomingMessages(requests, responses); |
| 666 | } |
| 667 | |
| 668 | void _processCDCMessages() { |
| 669 | int startUpdateSize = _updateSize(); |
no test coverage detected