| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| #if WITH_WS_VERSION |
| #include <ws_version.h> |
| constexpr int kWiresharkMajorVersion = WIRESHARK_VERSION_MAJOR; |
| constexpr int kWiresharkMinorVersion = WIRESHARK_VERSION_MINOR; |
| #else |
| #include <config.h> |
| constexpr int kWiresharkMajorVersion = VERSION_MAJOR; |
| constexpr int kWiresharkMinorVersion = VERSION_MINOR; |
| #endif |
| |
| #include <epan/column-utils.h> |
| #include <epan/dissectors/packet-tcp.h> |
| #include <epan/packet.h> |
| #include <epan/proto.h> |
| #include <epan/value_string.h> |
| #include <glib.h> |
| #include <wsutil/nstime.h> |
| |
| #include <map> |
| |
| #include "PulsarApi.pb.h" |
| |
| #ifdef VERSION |
| #undef VERSION |
| #endif |
| |
| /* Version number of package */ |
| #define VERSION "0.0.1" |
| |
| const static int PULSAR_PORT = 6650; |
| |
| static int proto_pulsar = -1; |
| static int hf_pulsar_error = -1; |
| static int hf_pulsar_error_message = -1; |
| static int hf_pulsar_cmd_type = -1; |
| static int hf_pulsar_frame_size = -1; |
| static int hf_pulsar_cmd_size = -1; |
| |
| static int hf_pulsar_client_version = -1; |
| static int hf_pulsar_auth_method = -1; |
| static int hf_pulsar_auth_data = -1; |
| static int hf_pulsar_protocol_version = -1; |
| static int hf_pulsar_server_version = -1; |
| |
| static int hf_pulsar_topic = -1; |
| static int hf_pulsar_subscription = -1; |
| static int hf_pulsar_subType = -1; |
| static int hf_pulsar_consumer_id = -1; |
| static int hf_pulsar_producer_id = -1; |
| static int hf_pulsar_server_error = -1; |
| static int hf_pulsar_ack_type = -1; |
| static int hf_pulsar_request_id = -1; |
| static int hf_pulsar_consumer_name = -1; |
| static int hf_pulsar_producer_name = -1; |
| |
| static int hf_pulsar_publish_time = -1; |
| static int hf_pulsar_deliver_at_time = -1; |
| static int hf_pulsar_event_time = -1; |
| static int hf_pulsar_deliver_after_time = -1; |
| |
| static int hf_pulsar_chunk_id = -1; |
| static int hf_pulsar_num_chunks_from_msg = -1; |
| static int hf_pulsar_uuid = -1; |
| static int hf_pulsar_compression_type = -1; |
| static int hf_pulsar_uncompressed_size = -1; |
| static int hf_pulsar_partition_key = -1; |
| static int hf_pulsar_ordering_key = -1; |
| static int hf_pulsar_encryption_algo = -1; |
| static int hf_pulsar_encryption_param = -1; |
| static int hf_pulsar_encryption_keys = -1; |
| static int hf_pulsar_replicated_from = -1; |
| static int hf_pulsar_replicate_to = -1; |
| static int hf_pulsar_property = -1; |
| |
| static int hf_pulsar_txnid_least_bits = -1; |
| static int hf_pulsar_txnid_most_bits = -1; |
| |
| static int hf_pulsar_request_in = -1; |
| static int hf_pulsar_response_in = -1; |
| static int hf_pulsar_publish_latency = -1; |
| |
| static int hf_pulsar_sequence_id = -1; |
| static int hf_pulsar_highest_sequence_id = -1; |
| static int hf_pulsar_message_id = -1; |
| static int hf_pulsar_message_permits = -1; |
| |
| static int ett_pulsar = -1; |
| |
| const static int FRAME_SIZE_LEN = 4; |
| const static guint16 MAGIC_BROKER_ENTRY_METADATA = 0x0e02; |
| const static guint16 MAGIC_CRC32C = 0x0e01; |
| |
| static pulsar::proto::BaseCommand command; |
| |
| using namespace pulsar::proto; |
| |
| static const value_string pulsar_cmd_names[] = { |
| {BaseCommand::CONNECT, "Connect"}, |
| {BaseCommand::CONNECTED, "Connected"}, |
| {BaseCommand::SUBSCRIBE, "Subscribe"}, |
| {BaseCommand::PRODUCER, "Producer"}, |
| {BaseCommand::SEND, "Send"}, |
| {BaseCommand::SEND_RECEIPT, "SendReceipt"}, |
| {BaseCommand::SEND_ERROR, "SendError"}, |
| {BaseCommand::MESSAGE, "Message"}, |
| {BaseCommand::ACK, "Ack"}, |
| {BaseCommand::FLOW, "Flow"}, |
| {BaseCommand::UNSUBSCRIBE, "Unsubscribe"}, |
| {BaseCommand::SUCCESS, "Success"}, |
| {BaseCommand::ERROR, "Error"}, |
| {BaseCommand::CLOSE_PRODUCER, "CloseProducer"}, |
| {BaseCommand::CLOSE_CONSUMER, "CloseConsumer"}, |
| {BaseCommand::PRODUCER_SUCCESS, "ProducerSuccess"}, |
| {BaseCommand::PING, "Ping"}, |
| {BaseCommand::PONG, "Pong"}, |
| {BaseCommand::REDELIVER_UNACKNOWLEDGED_MESSAGES, "RedeliverUnacknowledgedMessages"}, |
| {BaseCommand::PARTITIONED_METADATA, "PartitionedMetadata"}, |
| {BaseCommand::PARTITIONED_METADATA_RESPONSE, "PartitionedMetadataResponse"}, |
| {BaseCommand::LOOKUP, "Lookup"}, |
| {BaseCommand::LOOKUP_RESPONSE, "LookupResponse"}, |
| {BaseCommand::CONSUMER_STATS, "ConsumerStats"}, |
| {BaseCommand::CONSUMER_STATS_RESPONSE, "ConsumerStatsResponse"}, |
| {BaseCommand::SEEK, "Seek"}, |
| {BaseCommand::GET_LAST_MESSAGE_ID, "GetLastMessageId"}, |
| {BaseCommand::GET_LAST_MESSAGE_ID_RESPONSE, "GetLastMessageIdResponse"}, |
| {BaseCommand::ACTIVE_CONSUMER_CHANGE, "ActiveConsumerChange"}, |
| {BaseCommand::GET_TOPICS_OF_NAMESPACE, "GetTopicsOfNamespace"}, |
| {BaseCommand::GET_TOPICS_OF_NAMESPACE_RESPONSE, "GetTopicsOfNamespaceResponse"}, |
| {BaseCommand::GET_SCHEMA, "GetSchema"}, |
| {BaseCommand::GET_SCHEMA_RESPONSE, "GetSchemaResponse"}, |
| {BaseCommand::AUTH_CHALLENGE, "AuthChallenge"}, |
| {BaseCommand::AUTH_RESPONSE, "AuthResponse"}, |
| {BaseCommand::ACK_RESPONSE, "AckResponse"}, |
| {BaseCommand::GET_OR_CREATE_SCHEMA, "GetOrCreateSchema"}, |
| {BaseCommand::AUTH_CHALLENGE, "AuthChallenge"}, |
| {BaseCommand::AUTH_RESPONSE, "AuthResponse"}, |
| {BaseCommand::ACK_RESPONSE, "AckResponse"}, |
| {BaseCommand::GET_OR_CREATE_SCHEMA, "GetOrCreateSchema"}, |
| {BaseCommand::GET_OR_CREATE_SCHEMA_RESPONSE, "GetOrCreateSchemaResponse"}, |
| {BaseCommand::NEW_TXN, "NewTxn"}, |
| {BaseCommand::NEW_TXN_RESPONSE, "NewTxnResponse"}, |
| {BaseCommand::ADD_PARTITION_TO_TXN, "AddPartitionToTxn"}, |
| {BaseCommand::ADD_SUBSCRIPTION_TO_TXN, "AddSubscriptionToTxn"}, |
| {BaseCommand::ADD_SUBSCRIPTION_TO_TXN_RESPONSE, "AddSubscriptionToTxnResponse"}, |
| {BaseCommand::END_TXN, "EndTxn"}, |
| {BaseCommand::END_TXN_RESPONSE, "EndTxnResponse"}, |
| {BaseCommand::END_TXN_ON_PARTITION, "EndTxnOnPartition"}, |
| {BaseCommand::END_TXN_ON_PARTITION_RESPONSE, "EndTxnOnPartitionResponse"}, |
| {BaseCommand::END_TXN_ON_SUBSCRIPTION, "EndTxnOnSubscription"}, |
| {BaseCommand::END_TXN_ON_SUBSCRIPTION_RESPONSE, "EndTxnOnSubscriptionResponse"}, |
| {BaseCommand::TC_CLIENT_CONNECT_REQUEST, "TcClientConnectRequest"}, |
| {BaseCommand::TC_CLIENT_CONNECT_RESPONSE, "TcClientConnectResponse"}, |
| }; |
| |
| static const value_string auth_methods_vs[] = { |
| {AuthMethodNone, "None"}, // |
| {AuthMethodYcaV1, "YCAv1"}, // |
| {AuthMethodAthens, "Athens"} // |
| }; |
| |
| static const value_string server_errors_vs[] = { |
| {UnknownError, "UnknownError"}, |
| {MetadataError, "MetadataError"}, |
| {PersistenceError, "PersistenceError"}, |
| {AuthenticationError, "AuthenticationError"}, |
| {AuthorizationError, "AuthorizationError"}, |
| {ConsumerBusy, "ConsumerBusy"}, |
| {ServiceNotReady, "ServiceNotReady"}, |
| {ProducerBlockedQuotaExceededError, "ProducerBlockedQuotaExceededError"}, |
| {ProducerBlockedQuotaExceededException, "ProducerBlockedQuotaExceededException"}, |
| {ChecksumError, "ChecksumError"}, |
| {UnsupportedVersionError, "UnsupportedVersionError"}, |
| {TopicNotFound, "TopicNotFound"}, |
| {SubscriptionNotFound, "SubscriptionNotFound"}, |
| {ConsumerNotFound, "ConsumerNotFound"}, |
| {TooManyRequests, "TooManyRequests"}, |
| {TopicTerminatedError, "TopicTerminatedError"}, |
| {ProducerBusy, "ProducerBusy"}, |
| {InvalidTopicName, "InvalidTopicName"}, |
| {IncompatibleSchema, "IncompatibleSchema"}, |
| {ConsumerAssignError, "ConsumerAssignError"}, |
| {TransactionCoordinatorNotFound, "TransactionCoordinatorNotFound"}, |
| {InvalidTxnStatus, "InvalidTxnStatus"}, |
| {NotAllowedError, "NotAllowedError"}, |
| {TransactionConflict, "TransactionConflict"}, |
| {TransactionNotFound, "TransactionNotFound"}, |
| {ProducerFenced, "ProducerFenced"}, |
| }; |
| |
| static const value_string ack_type_vs[] = {{CommandAck::Individual, "Individual"}, |
| {CommandAck::Cumulative, "Cumulative"}}; |
| |
| static const value_string protocol_version_vs[] = { |
| {v0, "v0"}, {v1, "v1"}, {v2, "v2"}, {v3, "v3"}, {v4, "v4"}, {v5, "v5"}, {v6, "v6"}, |
| {v7, "v7"}, {v8, "v8"}, {v9, "v9"}, {v10, "v10"}, {v11, "v11"}, {v12, "v12"}, {v13, "v13"}, |
| {v14, "v14"}, {v15, "v15"}, {v16, "v16"}, {v17, "v17"}, {v18, "v18"}, {v19, "v19"}, |
| }; |
| |
| static const value_string sub_type_names_vs[] = { |
| {CommandSubscribe::Exclusive, "Exclusive"}, |
| {CommandSubscribe::Shared, "Shared"}, |
| {CommandSubscribe::Failover, "Failover"}, |
| {CommandSubscribe::Key_Shared, "Key_Shared"}, |
| }; |
| |
| static const value_string compression_type_name_vs[] = { |
| {CompressionType::NONE, "NONE"}, {CompressionType::LZ4, "LZ4"}, {CompressionType::ZLIB, "ZLIB"}, |
| {CompressionType::ZSTD, "ZSTD"}, {CompressionType::SNAPPY, "SNAPPY"}, |
| }; |
| |
| static const char* to_str(int value, const value_string* values) { |
| return val_to_str(value, values, "Unknown (%d)"); |
| } |
| |
| struct MessageIdComparator { |
| bool operator()(const MessageIdData& a, const MessageIdData& b) const { |
| if (a.ledgerid() < b.ledgerid()) { |
| return true; |
| } else if (a.ledgerid() == b.ledgerid()) { |
| return a.entryid() < b.entryid(); |
| } else { |
| return false; |
| } |
| } |
| }; |
| |
| struct RequestData { |
| uint32_t requestFrame; |
| nstime_t requestTimestamp; |
| uint32_t ackFrame; |
| nstime_t ackTimestamp; |
| |
| RequestData() : requestFrame(UINT32_MAX), ackFrame(UINT32_MAX) {} |
| }; |
| |
| struct RequestResponseData : public RequestData { |
| uint64_t id; // producer / consumer id |
| }; |
| |
| struct ProducerData { |
| std::string topic; |
| std::string producerName; |
| |
| std::map<uint64_t, RequestData> messages; |
| }; |
| |
| struct ConsumerData { |
| std::string topic; |
| std::string subscriptionName; |
| std::string consumerName; |
| CommandSubscribe::SubType subType; |
| |
| // Link messages to acks for the consumer |
| std::map<MessageIdData, RequestData, MessageIdComparator> messages; |
| }; |
| |
| struct ConnectionState { |
| std::map<uint64_t, ProducerData> producers; |
| std::map<uint64_t, ConsumerData> consumers; |
| std::map<uint64_t, RequestResponseData> requests; |
| }; |
| |
| static void dissect_message_metadata(proto_tree* frame_tree, tvbuff_t* tvb, int offset, int maxOffset) { |
| if (tvb_get_ntohs(tvb, offset) == MAGIC_BROKER_ENTRY_METADATA) { |
| offset += 2; |
| auto brokerEntryMetadataSize = (int)tvb_get_ntohl(tvb, offset); |
| offset += brokerEntryMetadataSize + 2; |
| #ifdef DEBUG |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] MAGIC_BROKER_ENTRY_METADATA %d", |
| brokerEntryMetadataSize); |
| #endif |
| } |
| if (tvb_get_ntohs(tvb, offset) == MAGIC_CRC32C) { |
| #ifdef DEBUG |
| auto checksum = tvb_get_ntohl(tvb, offset); |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] CRC32C %d", checksum); |
| #endif |
| // Skip CRC32C Magic (2) and CRC32C checksum (4) |
| offset += 2; |
| offset += 4; |
| } |
| // Decode message metadata |
| auto metadataSize = tvb_get_ntohl(tvb, offset); |
| offset += 4; |
| #ifdef DEBUG |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] MetadataSize %d, maxOffset %d", metadataSize, maxOffset); |
| #endif |
| |
| if (offset + metadataSize > maxOffset) { |
| // Not enough data to dissect metadata |
| #ifdef DEBUG |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] Not enough data to dissect message metadata"); |
| #endif |
| return; |
| } |
| |
| static MessageMetadata msgMetadata; |
| auto ptr = tvb_get_ptr(tvb, offset, metadataSize); |
| |
| if (!msgMetadata.ParseFromArray(ptr, metadataSize)) { |
| proto_tree_add_boolean_format(frame_tree, hf_pulsar_error, tvb, offset, metadataSize, true, |
| "Error parsing protocol buffer message metadata"); |
| return; |
| } |
| |
| #ifdef DEBUG |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] MessageMetadata Utf8DebugString : %s", |
| msgMetadata.Utf8DebugString().c_str()); |
| proto_tree_add_debug_text(frame_tree, "[DEBUG] MessageMetadata SerializeAsString : %s", |
| msgMetadata.SerializeAsString().c_str()); |
| #endif |
| |
| proto_item* md_tree = |
| proto_tree_add_subtree_format(frame_tree, tvb, offset, metadataSize, ett_pulsar, nullptr, |
| "MessageMetadata / %s / %" G_GINT64_MODIFIER "u", |
| msgMetadata.producer_name().c_str(), msgMetadata.sequence_id()); |
| proto_tree_add_string(md_tree, hf_pulsar_producer_name, tvb, offset, metadataSize, |
| msgMetadata.producer_name().c_str()); |
| |
| // IDs |
| proto_tree_add_uint64(md_tree, hf_pulsar_sequence_id, tvb, offset, metadataSize, |
| msgMetadata.sequence_id()); |
| if (msgMetadata.has_highest_sequence_id()) { |
| proto_tree_add_uint64(md_tree, hf_pulsar_highest_sequence_id, tvb, offset, metadataSize, |
| msgMetadata.highest_sequence_id()); |
| } |
| |
| if (msgMetadata.has_chunk_id()) { |
| proto_tree_add_uint(md_tree, hf_pulsar_chunk_id, tvb, offset, metadataSize, msgMetadata.chunk_id()); |
| if (msgMetadata.has_num_chunks_from_msg()) { |
| proto_tree_add_uint(md_tree, hf_pulsar_num_chunks_from_msg, tvb, offset, metadataSize, |
| msgMetadata.num_chunks_from_msg()); |
| } |
| } |
| |
| if (msgMetadata.has_uuid()) { |
| proto_tree_add_string(md_tree, hf_pulsar_uuid, tvb, offset, metadataSize, msgMetadata.uuid().c_str()); |
| } |
| |
| // Times |
| proto_tree_add_uint64(md_tree, hf_pulsar_publish_time, tvb, offset, metadataSize, |
| msgMetadata.publish_time()); |
| |
| if (msgMetadata.has_deliver_at_time()) { |
| proto_tree_add_uint64(md_tree, hf_pulsar_deliver_at_time, tvb, offset, metadataSize, |
| msgMetadata.deliver_at_time()); |
| proto_tree_add_uint64(md_tree, hf_pulsar_deliver_after_time, tvb, offset, metadataSize, |
| msgMetadata.deliver_at_time() - msgMetadata.publish_time()); |
| } |
| |
| if (msgMetadata.has_event_time()) { |
| proto_tree_add_uint64(md_tree, hf_pulsar_event_time, tvb, offset, metadataSize, |
| msgMetadata.event_time()); |
| } |
| |
| // Compression |
| if (msgMetadata.has_compression()) { |
| proto_tree_add_string(md_tree, hf_pulsar_compression_type, tvb, offset, metadataSize, |
| to_str(msgMetadata.compression(), compression_type_name_vs)); |
| } |
| |
| if (msgMetadata.has_uncompressed_size()) { |
| proto_tree_add_uint(md_tree, hf_pulsar_uncompressed_size, tvb, offset, metadataSize, |
| msgMetadata.uncompressed_size()); |
| } |
| |
| // Keys |
| if (msgMetadata.has_partition_key()) { |
| proto_tree_add_string(md_tree, hf_pulsar_partition_key, tvb, offset, metadataSize, |
| msgMetadata.partition_key().c_str()); |
| } |
| |
| if (msgMetadata.has_ordering_key()) { |
| proto_tree_add_string(md_tree, hf_pulsar_ordering_key, tvb, offset, metadataSize, |
| msgMetadata.ordering_key().c_str()); |
| } |
| |
| // Encryption |
| if (msgMetadata.has_encryption_algo()) { |
| proto_tree_add_string(md_tree, hf_pulsar_encryption_algo, tvb, offset, metadataSize, |
| msgMetadata.encryption_algo().c_str()); |
| } |
| if (msgMetadata.has_encryption_param()) { |
| proto_tree_add_string(md_tree, hf_pulsar_encryption_param, tvb, offset, metadataSize, |
| msgMetadata.encryption_param().c_str()); |
| } |
| if (msgMetadata.encryption_keys_size() > 0) { |
| proto_item* encryption_keys_tree = proto_tree_add_subtree_format( |
| md_tree, tvb, offset, msgMetadata.encryption_param().size(), ett_pulsar, nullptr, |
| "EncryptionParam / %s", msgMetadata.encryption_algo().c_str()); |
| for (int i = 0; i < msgMetadata.encryption_keys().size(); i++) { |
| const auto& encryption_key = msgMetadata.encryption_keys(i); |
| proto_tree_add_string_format( |
| encryption_keys_tree, hf_pulsar_encryption_keys, tvb, offset, metadataSize, "", "%s : %s", |
| encryption_key.has_key() ? encryption_key.key().c_str() : "<none>", |
| encryption_key.has_value() ? encryption_key.value().c_str() : "<none>"); |
| } |
| } |
| |
| // Properties |
| if (msgMetadata.properties_size() > 0) { |
| proto_item* properties_tree = proto_tree_add_subtree_format(md_tree, tvb, offset, metadataSize, |
| ett_pulsar, nullptr, "Properties"); |
| for (int i = 0; i < msgMetadata.properties_size(); i++) { |
| const KeyValue& kv = msgMetadata.properties(i); |
| proto_tree_add_string_format(properties_tree, hf_pulsar_property, tvb, offset, metadataSize, "", |
| "%s : %s", kv.key().c_str(), kv.value().c_str()); |
| } |
| } |
| |
| // Replication |
| if (msgMetadata.has_replicated_from()) { |
| proto_tree_add_string(md_tree, hf_pulsar_replicated_from, tvb, offset, metadataSize, |
| msgMetadata.replicated_from().c_str()); |
| } |
| |
| if (msgMetadata.replicate_to_size() > 0) { |
| proto_item* replicate_tree = proto_tree_add_subtree_format(md_tree, tvb, offset, metadataSize, |
| ett_pulsar, nullptr, "Replicate to"); |
| for (int i = 0; i < msgMetadata.replicate_to_size(); i++) { |
| proto_tree_add_string_format(replicate_tree, hf_pulsar_replicated_from, tvb, offset, metadataSize, |
| "", "%s", msgMetadata.replicate_to(i).c_str()); |
| } |
| } |
| |
| // Transaction |
| if (msgMetadata.has_txnid_least_bits()) { |
| proto_tree_add_uint64(md_tree, hf_pulsar_txnid_least_bits, tvb, offset, metadataSize, |
| msgMetadata.txnid_least_bits()); |
| } |
| |
| if (msgMetadata.has_txnid_most_bits()) { |
| proto_tree_add_uint64(md_tree, hf_pulsar_txnid_most_bits, tvb, offset, metadataSize, |
| msgMetadata.txnid_most_bits()); |
| } |
| |
| // Payloads |
| offset += metadataSize; |
| uint32_t payloadSize = maxOffset - offset; |
| proto_tree_add_subtree_format(md_tree, tvb, offset, payloadSize, ett_pulsar, nullptr, "Payload / size=%u", |
| payloadSize); |
| } |
| |
| void link_to_request_frame(proto_tree* cmd_tree, tvbuff_t* tvb, int offset, int size, |
| const RequestData& reqData) { |
| if (reqData.requestFrame != UINT32_MAX) { |
| proto_tree* item = |
| proto_tree_add_uint(cmd_tree, hf_pulsar_request_in, tvb, offset, size, reqData.requestFrame); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| nstime_t latency; |
| nstime_delta(&latency, &reqData.ackTimestamp, &reqData.requestTimestamp); |
| item = proto_tree_add_time(cmd_tree, hf_pulsar_publish_latency, tvb, offset, size, &latency); |
| PROTO_ITEM_SET_GENERATED(item); |
| } |
| } |
| |
| void link_to_response_frame(proto_tree* cmd_tree, tvbuff_t* tvb, int offset, int size, |
| const RequestData& reqData) { |
| if (reqData.ackFrame != UINT32_MAX) { |
| proto_tree* item = |
| proto_tree_add_uint(cmd_tree, hf_pulsar_response_in, tvb, offset, size, reqData.ackFrame); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| nstime_t latency; |
| nstime_delta(&latency, &reqData.ackTimestamp, &reqData.requestTimestamp); |
| item = proto_tree_add_time(cmd_tree, hf_pulsar_publish_latency, tvb, offset, size, &latency); |
| PROTO_ITEM_SET_GENERATED(item); |
| } |
| } |
| |
| ////////// |
| |
| /* This method dissects fully reassembled messages */ |
| static int dissect_pulsar_message(tvbuff_t* tvb, packet_info* pinfo, proto_tree* tree, void* data _U_) { |
| uint32_t offset = FRAME_SIZE_LEN; |
| int maxOffset = tvb_captured_length(tvb); |
| auto cmdSize = (uint32_t)tvb_get_ntohl(tvb, offset); |
| offset += 4; |
| |
| if (offset + cmdSize > maxOffset) { |
| // Not enough data to dissect |
| #ifdef DEBUG |
| proto_tree_add_debug_text(tree, "[Debug] Not enough data to dissect command"); |
| #endif |
| return maxOffset; |
| } |
| |
| col_set_str(pinfo->cinfo, COL_PROTOCOL, "Pulsar"); |
| |
| conversation_t* conversation = find_or_create_conversation(pinfo); |
| auto state = (ConnectionState*)conversation_get_proto_data(conversation, proto_pulsar); |
| if (state == nullptr) { |
| state = new ConnectionState(); |
| conversation_add_proto_data(conversation, proto_pulsar, state); |
| } |
| |
| auto ptr = (uint8_t*)tvb_get_ptr(tvb, offset, cmdSize); |
| if (!command.ParseFromArray(ptr, cmdSize)) { |
| proto_tree_add_boolean_format(tree, hf_pulsar_error, tvb, offset, cmdSize, true, |
| "Error parsing protocol buffer command"); |
| return maxOffset; |
| } |
| |
| int cmdOffset = offset; |
| offset += cmdSize; |
| |
| col_add_str(pinfo->cinfo, COL_INFO, to_str(command.type(), pulsar_cmd_names)); |
| |
| proto_item* frame_tree = nullptr; |
| proto_item* cmd_tree = nullptr; |
| if (tree) { /* we are being asked for details */ |
| proto_item* ti = proto_tree_add_item(tree, proto_pulsar, tvb, 0, -1, ENC_NA); |
| frame_tree = proto_item_add_subtree(ti, ett_pulsar); |
| proto_tree_add_item(frame_tree, hf_pulsar_frame_size, tvb, 0, 4, ENC_BIG_ENDIAN); |
| proto_tree_add_item(frame_tree, hf_pulsar_cmd_size, tvb, 4, 4, ENC_BIG_ENDIAN); |
| cmd_tree = proto_tree_add_subtree_format(frame_tree, tvb, 8, cmdSize, ett_pulsar, nullptr, |
| "Command %s", to_str(command.type(), pulsar_cmd_names)); |
| proto_tree_add_string(cmd_tree, hf_pulsar_cmd_type, tvb, 8, cmdSize, |
| to_str(command.type(), pulsar_cmd_names)); |
| } |
| |
| switch (command.type()) { |
| case BaseCommand::CONNECT: { |
| const CommandConnect& connect = command.connect(); |
| if (tree) { |
| proto_tree_add_string(cmd_tree, hf_pulsar_client_version, tvb, cmdOffset, cmdSize, |
| connect.client_version().c_str()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_protocol_version, tvb, cmdOffset, cmdSize, |
| to_str(connect.protocol_version(), protocol_version_vs)); |
| proto_tree_add_string(cmd_tree, hf_pulsar_auth_method, tvb, cmdOffset, cmdSize, |
| to_str(connect.auth_method(), auth_methods_vs)); |
| if (connect.has_auth_data()) { |
| proto_tree_add_string(cmd_tree, hf_pulsar_auth_data, tvb, cmdOffset, cmdSize, |
| connect.auth_data().c_str()); |
| } |
| } |
| break; |
| } |
| case BaseCommand::CONNECTED: { |
| const CommandConnected& connected = command.connected(); |
| if (tree) { |
| proto_tree_add_string(cmd_tree, hf_pulsar_server_version, tvb, cmdOffset, cmdSize, |
| connected.server_version().c_str()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_protocol_version, tvb, cmdOffset, cmdSize, |
| to_str(connected.protocol_version(), protocol_version_vs)); |
| } |
| break; |
| } |
| case BaseCommand::SUBSCRIBE: { |
| const CommandSubscribe& subscribe = command.subscribe(); |
| RequestData& reqData = state->requests[subscribe.request_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ConsumerData& consumerData = state->consumers[subscribe.consumer_id()]; |
| consumerData.topic = subscribe.topic(); |
| consumerData.subscriptionName = subscribe.subscription(); |
| consumerData.consumerName = subscribe.consumer_name(); |
| consumerData.subType = subscribe.subtype(); |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %s / %s / %s", |
| to_str(subscribe.subtype(), sub_type_names_vs), subscribe.topic().c_str(), |
| subscribe.subscription().c_str(), subscribe.consumer_name().c_str()); |
| |
| if (tree) { |
| proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| subscribe.topic().c_str()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_subscription, tvb, cmdOffset, cmdSize, |
| subscribe.subscription().c_str()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_subType, tvb, cmdOffset, cmdSize, |
| to_str(subscribe.subtype(), sub_type_names_vs)); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| subscribe.consumer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| subscribe.request_id()); |
| proto_tree_add_string( |
| cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| subscribe.has_consumer_name() ? subscribe.consumer_name().c_str() : "<none>"); |
| } |
| break; |
| } |
| case BaseCommand::PRODUCER: { |
| const CommandProducer& producer = command.producer(); |
| RequestResponseData& reqData = state->requests[producer.request_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| reqData.id = producer.producer_id(); |
| |
| state->producers[producer.producer_id()].topic = producer.topic(); |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s", producer.topic().c_str()); |
| |
| if (tree) { |
| proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producer.topic().c_str()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| producer.producer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| producer.request_id()); |
| proto_tree_add_string( |
| cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| producer.has_producer_name() ? producer.producer_name().c_str() : "<none>"); |
| |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::SEND: { |
| const CommandSend& send = command.send(); |
| RequestData& reqData = state->producers[send.producer_id()].messages[send.sequence_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ProducerData& producerData = state->producers[send.producer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %" G_GINT64_MODIFIER "u", |
| producerData.producerName.c_str(), send.sequence_id()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| send.producer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_sequence_id, tvb, cmdOffset, cmdSize, |
| send.sequence_id()); |
| |
| // Decode message metadata |
| dissect_message_metadata(cmd_tree, tvb, offset, maxOffset); |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| producerData.producerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| // Pair with frame information |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::SEND_RECEIPT: { |
| const CommandSendReceipt& send_receipt = command.send_receipt(); |
| RequestData& reqData = |
| state->producers[send_receipt.producer_id()].messages[send_receipt.sequence_id()]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ProducerData& producerData = state->producers[send_receipt.producer_id()]; |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %" G_GINT64_MODIFIER "u", |
| producerData.producerName.c_str(), send_receipt.sequence_id()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| send_receipt.producer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_sequence_id, tvb, cmdOffset, cmdSize, |
| send_receipt.sequence_id()); |
| if (send_receipt.has_message_id()) { |
| const MessageIdData& messageId = send_receipt.message_id(); |
| proto_tree_add_string_format(cmd_tree, hf_pulsar_message_id, tvb, cmdOffset, cmdSize, "", |
| "Message Id: %" G_GINT64_MODIFIER "u:%" G_GINT64_MODIFIER |
| "u", |
| messageId.ledgerid(), messageId.entryid()); |
| } |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| producerData.producerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::SEND_ERROR: { |
| const CommandSendError& send_error = command.send_error(); |
| RequestData& reqData = |
| state->producers[send_error.producer_id()].messages[send_error.sequence_id()]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ProducerData& producerData = state->producers[send_error.producer_id()]; |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %" G_GINT64_MODIFIER "u", |
| producerData.producerName.c_str(), send_error.sequence_id()); |
| |
| if (tree) { |
| proto_tree_add_boolean_format(frame_tree, hf_pulsar_error, tvb, cmdOffset, cmdSize, true, |
| "Error in sending operation"); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| send_error.producer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_sequence_id, tvb, cmdOffset, cmdSize, |
| send_error.sequence_id()); |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_server_error, tvb, cmdOffset, cmdSize, |
| to_str(send_error.error(), server_errors_vs)); |
| |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| producerData.producerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| // Pair with frame information |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::MESSAGE: { |
| const CommandMessage& message = command.message(); |
| state->consumers[message.consumer_id()].messages[message.message_id()]; |
| RequestData& reqData = state->consumers[message.consumer_id()].messages[message.message_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| const ConsumerData& consumerData = state->consumers[message.consumer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %" G_GINT64_MODIFIER "u:%" G_GINT64_MODIFIER "u", |
| consumerData.consumerName.c_str(), message.message_id().ledgerid(), |
| message.message_id().entryid()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| message.consumer_id()); |
| |
| dissect_message_metadata(cmd_tree, tvb, offset, maxOffset); |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| consumerData.consumerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| consumerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| // Pair with frame information |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::ACK: { |
| const CommandAck& ack = command.ack(); |
| RequestData& reqData = state->consumers[ack.consumer_id()].messages[ack.message_id().Get(0)]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| const ConsumerData& consumerData = state->consumers[ack.consumer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %" G_GINT64_MODIFIER "u:%" G_GINT64_MODIFIER "u", |
| consumerData.consumerName.c_str(), ack.message_id().Get(0).ledgerid(), |
| ack.message_id().Get(0).entryid()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| ack.consumer_id()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_ack_type, tvb, cmdOffset, cmdSize, |
| to_str(ack.ack_type(), ack_type_vs)); |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| consumerData.consumerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| consumerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| // Pair with frame information |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::FLOW: { |
| const CommandFlow& flow = command.flow(); |
| const ConsumerData& consumerData = state->consumers[flow.consumer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %d", consumerData.consumerName.c_str(), |
| flow.messagepermits()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| flow.consumer_id()); |
| proto_tree_add_uint(cmd_tree, hf_pulsar_message_permits, tvb, cmdOffset, cmdSize, |
| flow.messagepermits()); |
| |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| consumerData.consumerName.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| consumerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| } |
| break; |
| } |
| case BaseCommand::UNSUBSCRIBE: { |
| const CommandUnsubscribe& unsubscribe = command.unsubscribe(); |
| RequestData& reqData = state->requests[unsubscribe.request_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ConsumerData& consumerData = state->consumers[unsubscribe.consumer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %s / %s / %s", |
| to_str(consumerData.subType, sub_type_names_vs), consumerData.topic.c_str(), |
| consumerData.subscriptionName.c_str(), consumerData.consumerName.c_str()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| unsubscribe.consumer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| unsubscribe.request_id()); |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| consumerData.topic.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_subscription, tvb, cmdOffset, cmdSize, |
| consumerData.subscriptionName.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| proto_tree_add_string(cmd_tree, hf_pulsar_subType, tvb, cmdOffset, cmdSize, |
| to_str(consumerData.subType, sub_type_names_vs)); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| proto_tree_add_string(cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| consumerData.consumerName.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| |
| case BaseCommand::SUCCESS: { |
| const CommandSuccess& success = command.success(); |
| RequestResponseData& reqData = state->requests[success.request_id()]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| success.request_id()); |
| |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::ERROR: { |
| const CommandError& error = command.error(); |
| RequestResponseData& reqData = state->requests[error.request_id()]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| if (tree) { |
| proto_tree_add_boolean_format(frame_tree, hf_pulsar_error, tvb, cmdOffset, cmdSize, true, |
| "Request failed"); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| error.request_id()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_server_error, tvb, cmdOffset, cmdSize, |
| to_str(error.error(), server_errors_vs)); |
| proto_tree_add_string(cmd_tree, hf_pulsar_error_message, tvb, cmdOffset, cmdSize, |
| error.message().c_str()); |
| |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::CLOSE_PRODUCER: { |
| const CommandCloseProducer& close_producer = command.close_producer(); |
| RequestData& reqData = state->requests[close_producer.request_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ProducerData& producerData = state->producers[close_producer.producer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s", producerData.topic.c_str()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| close_producer.producer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| close_producer.request_id()); |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producerData.topic.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| proto_tree_add_string(cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| producerData.producerName.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| |
| case BaseCommand::CLOSE_CONSUMER: { |
| const CommandCloseConsumer& close_consumer = command.close_consumer(); |
| RequestData& reqData = state->requests[close_consumer.request_id()]; |
| reqData.requestFrame = pinfo->fd->num; |
| reqData.requestTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.requestTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| |
| ConsumerData& consumerData = state->consumers[close_consumer.consumer_id()]; |
| |
| col_append_fstr(pinfo->cinfo, COL_INFO, " / %s / %s / %s / %s", |
| to_str(consumerData.subType, sub_type_names_vs), consumerData.topic.c_str(), |
| consumerData.subscriptionName.c_str(), consumerData.consumerName.c_str()); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_consumer_id, tvb, cmdOffset, cmdSize, |
| close_consumer.consumer_id()); |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| close_consumer.request_id()); |
| auto item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| consumerData.topic.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_subscription, tvb, cmdOffset, cmdSize, |
| consumerData.subscriptionName.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| proto_tree_add_string(cmd_tree, hf_pulsar_subType, tvb, cmdOffset, cmdSize, |
| to_str(consumerData.subType, sub_type_names_vs)); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| proto_tree_add_string(cmd_tree, hf_pulsar_consumer_name, tvb, cmdOffset, cmdSize, |
| consumerData.consumerName.c_str()); |
| PROTO_ITEM_IS_GENERATED(item); |
| |
| link_to_response_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| |
| case BaseCommand::PRODUCER_SUCCESS: { |
| const CommandProducerSuccess& success = command.producer_success(); |
| RequestResponseData& reqData = state->requests[success.request_id()]; |
| reqData.ackFrame = pinfo->fd->num; |
| reqData.ackTimestamp.secs = pinfo->fd->abs_ts.secs; |
| reqData.ackTimestamp.nsecs = pinfo->fd->abs_ts.nsecs; |
| uint64_t producerId = reqData.id; |
| ProducerData& producerData = state->producers[producerId]; |
| producerData.producerName = success.producer_name(); |
| |
| if (tree) { |
| proto_tree_add_uint64(cmd_tree, hf_pulsar_request_id, tvb, cmdOffset, cmdSize, |
| success.request_id()); |
| proto_tree_add_string(cmd_tree, hf_pulsar_producer_name, tvb, cmdOffset, cmdSize, |
| success.producer_name().c_str()); |
| |
| auto item = proto_tree_add_uint64(cmd_tree, hf_pulsar_producer_id, tvb, cmdOffset, cmdSize, |
| producerId); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| item = proto_tree_add_string(cmd_tree, hf_pulsar_topic, tvb, cmdOffset, cmdSize, |
| producerData.topic.c_str()); |
| PROTO_ITEM_SET_GENERATED(item); |
| |
| link_to_request_frame(cmd_tree, tvb, cmdOffset, cmdSize, reqData); |
| } |
| break; |
| } |
| case BaseCommand::PING: |
| break; |
| case BaseCommand::PONG: |
| break; |
| case BaseCommand::REDELIVER_UNACKNOWLEDGED_MESSAGES: |
| break; |
| case BaseCommand::PARTITIONED_METADATA: |
| break; |
| case BaseCommand::PARTITIONED_METADATA_RESPONSE: |
| break; |
| case BaseCommand::LOOKUP: |
| break; |
| case BaseCommand::LOOKUP_RESPONSE: |
| break; |
| case BaseCommand::CONSUMER_STATS: |
| break; |
| case BaseCommand::CONSUMER_STATS_RESPONSE: |
| break; |
| case BaseCommand::REACHED_END_OF_TOPIC: |
| break; |
| case BaseCommand::SEEK: |
| break; |
| case BaseCommand::GET_LAST_MESSAGE_ID: |
| break; |
| case BaseCommand::GET_LAST_MESSAGE_ID_RESPONSE: |
| break; |
| case BaseCommand::ACTIVE_CONSUMER_CHANGE: |
| break; |
| case BaseCommand::GET_TOPICS_OF_NAMESPACE: |
| break; |
| case BaseCommand::GET_TOPICS_OF_NAMESPACE_RESPONSE: |
| break; |
| case BaseCommand::GET_SCHEMA: |
| break; |
| case BaseCommand::GET_SCHEMA_RESPONSE: |
| break; |
| case BaseCommand::AUTH_CHALLENGE: |
| break; |
| case BaseCommand::AUTH_RESPONSE: |
| break; |
| case BaseCommand::ACK_RESPONSE: |
| break; |
| case BaseCommand::GET_OR_CREATE_SCHEMA: |
| break; |
| case BaseCommand::GET_OR_CREATE_SCHEMA_RESPONSE: |
| break; |
| case BaseCommand::NEW_TXN: |
| break; |
| case BaseCommand::NEW_TXN_RESPONSE: |
| break; |
| case BaseCommand::ADD_PARTITION_TO_TXN: |
| break; |
| case BaseCommand::ADD_PARTITION_TO_TXN_RESPONSE: |
| break; |
| case BaseCommand::ADD_SUBSCRIPTION_TO_TXN: |
| break; |
| case BaseCommand::ADD_SUBSCRIPTION_TO_TXN_RESPONSE: |
| break; |
| case BaseCommand::END_TXN: |
| break; |
| case BaseCommand::END_TXN_RESPONSE: |
| break; |
| case BaseCommand::END_TXN_ON_PARTITION: |
| break; |
| case BaseCommand::END_TXN_ON_PARTITION_RESPONSE: |
| break; |
| case BaseCommand::END_TXN_ON_SUBSCRIPTION: |
| break; |
| case BaseCommand::END_TXN_ON_SUBSCRIPTION_RESPONSE: |
| break; |
| case BaseCommand::TC_CLIENT_CONNECT_REQUEST: |
| break; |
| case BaseCommand::TC_CLIENT_CONNECT_RESPONSE: |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST: |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST_SUCCESS: |
| break; |
| case BaseCommand::WATCH_TOPIC_UPDATE: |
| break; |
| case BaseCommand::WATCH_TOPIC_LIST_CLOSE: |
| break; |
| } |
| |
| return maxOffset; |
| } |
| |
| /* determine PDU length of protocol Pulsar */ |
| static uint32_t get_pulsar_message_len(packet_info* pinfo _U_, tvbuff_t* tvb, int offset, void* data _U_) { |
| auto len = (uint32_t)tvb_get_ntohl(tvb, offset); |
| return FRAME_SIZE_LEN + len; |
| } |
| |
| static int dissect_pulsar(tvbuff_t* tvb, packet_info* pinfo, proto_tree* tree, void* data _U_) { |
| tcp_dissect_pdus(tvb, pinfo, tree, 1, FRAME_SIZE_LEN, get_pulsar_message_len, dissect_pulsar_message, |
| data); |
| return tvb_captured_length(tvb); |
| } |
| |
| static hf_register_info hf[] = { |
| // |
| {&hf_pulsar_error, {"Error", "apache.pulsar.error", FT_BOOLEAN, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_error_message, |
| {"Message", "apache.pulsar.error_message", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_cmd_type, |
| {"Command Type", "apache.pulsar.cmd.type", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_frame_size, |
| {"Frame size", "apache.pulsar.frame_size", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_cmd_size, |
| {"Command size", "apache.pulsar.cmd_size", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_client_version, |
| {"Client version", "apache.pulsar.client_version", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_auth_method, |
| {"Auth method", "apache.pulsar.auth_method", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_auth_data, |
| {"Auth data", "apache.pulsar.auth_data", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_protocol_version, |
| {"Protocol version", "apache.pulsar.protocol_version", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, |
| |
| {&hf_pulsar_server_version, |
| {"Server version", "apache.pulsar.server_version", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_topic, {"Topic", "apache.pulsar.topic", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_subscription, |
| {"Subscription", "apache.pulsar.subscription", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_subType, |
| {"Subscription type:", "apache.pulsar.sub_type", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_consumer_id, |
| {"Consumer Id", "apache.pulsar.consumer_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_producer_id, |
| {"Producer Id", "apache.pulsar.producer_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_server_error, |
| {"Server error", "apache.pulsar.server_error", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_ack_type, {"Ack type", "apache.pulsar.ack_type", FT_STRING, 0, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_request_id, |
| {"Request Id", "apache.pulsar.request_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_consumer_name, |
| {"Consumer Name", "apache.pulsar.consumer_name", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_producer_name, |
| {"Producer Name", "apache.pulsar.producer_name", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_sequence_id, |
| {"Sequence Id", "apache.pulsar.sequence_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_highest_sequence_id, |
| {"Highest Sequence Id", "apache.pulsar.highest_sequence_id", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, |
| HFILL}}, // |
| {&hf_pulsar_uuid, {"UUID", "apache.pulsar.uuid", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_message_id, |
| {"Message Id", "apache.pulsar.message_id", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_message_permits, |
| {"Message Permits", "apache.pulsar.message_permits", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_publish_time, |
| {"Publish Time", "apache.pulsar.publish_time", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_deliver_at_time, |
| {"Deliver At Time", "apache.pulsar.deliver_at_time", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_event_time, |
| {"Event Time", "apache.pulsar.event_time", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_deliver_after_time, |
| {"Deliver After Time", "apache.pulsar.deliver_after_time", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, |
| HFILL}}, // |
| |
| {&hf_pulsar_chunk_id, |
| {"Chunk Id", "apache.pulsar.chunk_id", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_num_chunks_from_msg, |
| {"Num Chunks From Msg", "apache.pulsar.num_chunks_from_msg", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, |
| HFILL}}, // |
| {&hf_pulsar_compression_type, |
| {"Compression Type", "apache.pulsar.compression_type", FT_STRING, BASE_NONE, NULL, 0x0, NULL, |
| HFILL}}, // |
| {&hf_pulsar_uncompressed_size, |
| {"UnCompression Size", "apache.pulsar.uncompressed_size", FT_UINT32, BASE_DEC, NULL, 0x0, NULL, |
| HFILL}}, // |
| |
| {&hf_pulsar_replicated_from, |
| {"Replicated from", "apache.pulsar.replicated_from", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_partition_key, |
| {"Partition Key", "apache.pulsar.partition_key", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_ordering_key, |
| {"Ordering Key", "apache.pulsar.ordering_key", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_encryption_algo, |
| {"Encryption Algo", "apache.pulsar.encryption_algo", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_encryption_param, |
| {"Encryption Param", "apache.pulsar.encryption_param", FT_STRING, BASE_NONE, NULL, 0x0, NULL, |
| HFILL}}, // |
| |
| {&hf_pulsar_replicate_to, |
| {"Replicate to", "apache.pulsar.replicate_to", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_property, |
| {"Property", "apache.pulsar.property", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_encryption_keys, |
| {"Encryption Keys", "apache.pulsar.encryption_keys", FT_STRING, BASE_NONE, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_txnid_least_bits, |
| {"TxnId Least Bits", "apache.pulsar.txnid_least_bits", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| {&hf_pulsar_txnid_most_bits, |
| {"TxnId Most Bits", "apache.pulsar.txnid_most_bits", FT_UINT64, BASE_DEC, NULL, 0x0, NULL, HFILL}}, // |
| |
| {&hf_pulsar_request_in, |
| {"Request in frame", "apache.pulsar.request_in", FT_FRAMENUM, BASE_NONE, NULL, 0, |
| "This packet is a response to the packet with this number", HFILL}}, // |
| {&hf_pulsar_response_in, |
| {"Response in frame", "apache.pulsar.response_in", FT_FRAMENUM, BASE_NONE, NULL, 0, |
| "This packet will be responded in the packet with this number", HFILL}}, // |
| {&hf_pulsar_publish_latency, |
| {"Latency", "apache.pulsar.publish_latency", FT_RELATIVE_TIME, BASE_NONE, NULL, 0x0, |
| "How long time it took to ACK message", HFILL}}, |
| }; |
| |
| //////////////// |
| /// |
| void proto_register_pulsar() { |
| // register the new protocol, protocol fields, and subtrees |
| static dissector_handle_t pulsar_handle; |
| |
| proto_pulsar = proto_register_protocol("Pulsar Wire Protocol", /* name */ |
| "Apache Pulsar", /* short name */ |
| "apache.pulsar" /* abbrev */ |
| ); |
| |
| /* Setup protocol subtree array */ |
| static int* ett[] = {&ett_pulsar}; |
| |
| proto_register_field_array(proto_pulsar, hf, array_length(hf)); |
| proto_register_subtree_array(ett, array_length(ett)); |
| |
| pulsar_handle = create_dissector_handle(&dissect_pulsar, proto_pulsar); |
| dissector_add_uint("tcp.port", PULSAR_PORT, pulsar_handle); |
| register_postdissector(pulsar_handle); |
| } |
| |
| extern "C" { |
| |
| extern __attribute__((unused)) WS_DLL_PUBLIC_DEF const gchar plugin_version[] = VERSION; |
| extern __attribute__((unused)) WS_DLL_PUBLIC_DEF const int plugin_want_major = kWiresharkMajorVersion; |
| extern __attribute__((unused)) WS_DLL_PUBLIC_DEF const int plugin_want_minor = kWiresharkMinorVersion; |
| |
| WS_DLL_PUBLIC void plugin_register(void); |
| |
| __attribute__((unused)) static void proto_reg_handoff_pulsar(void) {} |
| |
| /* Start the functions we need for the plugin stuff */ |
| void plugin_register(void) { |
| static proto_plugin plug; |
| plug.register_protoinfo = proto_register_pulsar; |
| plug.register_handoff = proto_reg_handoff_pulsar; /* or nullptr */ |
| proto_register_plugin(&plug); |
| } |
| } |
