| 837 | |
| 838 | template <class T> |
| 839 | void LogPushData::writeTypedMessage(T const& item, bool metadataMessage, bool allLocations) { |
| 840 | prev_tags.clear(); |
| 841 | if (logSystem->hasRemoteLogs()) { |
| 842 | prev_tags.push_back(logSystem->getRandomRouterTag()); |
| 843 | } |
| 844 | for (auto& tag : next_message_tags) { |
| 845 | prev_tags.push_back(tag); |
| 846 | } |
| 847 | msg_locations.clear(); |
| 848 | logSystem->getPushLocations(prev_tags, msg_locations, allLocations); |
| 849 | |
| 850 | BinaryWriter bw(AssumeVersion(g_network->protocolVersion())); |
| 851 | |
| 852 | // Metadata messages (currently LogProtocolMessage is the only metadata |
| 853 | // message) should be written before span information. If this isn't a |
| 854 | // metadata message, make sure all locations have had transaction info |
| 855 | // written to them. Mutations may have different sets of tags, so it |
| 856 | // is necessary to check all tag locations each time a mutation is |
| 857 | // written. |
| 858 | if (!metadataMessage) { |
| 859 | uint32_t subseq = this->subsequence++; |
| 860 | bool updatedLocation = false; |
| 861 | for (int loc : msg_locations) { |
| 862 | updatedLocation = writeTransactionInfo(loc, subseq) || updatedLocation; |
| 863 | } |
| 864 | // If this message doesn't write to any new locations, the |
| 865 | // subsequence wasn't actually used and can be decremented. |
| 866 | if (!updatedLocation) { |
| 867 | this->subsequence--; |
| 868 | CODE_PROBE(true, "No new SpanContextMessage written to transaction logs"); |
| 869 | ASSERT(this->subsequence > 0); |
| 870 | } |
| 871 | } else { |
| 872 | // When writing a metadata message, make sure transaction state has |
| 873 | // been reset. If you are running into this assertion, make sure |
| 874 | // you are calling addTransactionInfo before each transaction. |
| 875 | ASSERT(writtenLocations.size() == 0); |
| 876 | } |
| 877 | |
| 878 | uint32_t subseq = this->subsequence++; |
| 879 | bool first = true; |
| 880 | int firstOffset = -1, firstLength = -1; |
| 881 | for (int loc : msg_locations) { |
| 882 | BinaryWriter& wr = messagesWriter[loc]; |
| 883 | |
| 884 | if (first) { |
| 885 | firstOffset = wr.getLength(); |
| 886 | wr << uint32_t(0) << subseq << uint16_t(prev_tags.size()); |
| 887 | for (auto& tag : prev_tags) |
| 888 | wr << tag; |
| 889 | wr << item; |
| 890 | firstLength = wr.getLength() - firstOffset; |
| 891 | *(uint32_t*)((uint8_t*)wr.getData() + firstOffset) = firstLength - sizeof(uint32_t); |
| 892 | DEBUG_TAGS_AND_MESSAGE( |
| 893 | "ProxyPushLocations", invalidVersion, StringRef(((uint8_t*)wr.getData() + firstOffset), firstLength)) |
| 894 | .detail("PushLocations", msg_locations); |
| 895 | first = false; |
| 896 | } else { |
no test coverage detected