diff options
| author | Piotr Smolinski <piotr.smolinski@confluent.io> | 2019-08-12 23:57:45 +0200 |
|---|---|---|
| committer | Anders Broman <a.broman58@gmail.com> | 2019-08-20 13:50:45 +0000 |
| commit | ad94c4d459d243c0cbbb9b222d5f7cdf8189ab86 (patch) | |
| tree | 06691a19b55777e1aad0a3f79c1ce6ceac0e9ae7 /epan | |
| parent | 8b8ce52abcf98f44bea025809c75f22224825ecb (diff) | |
Kafka: include zstd compression in Kafka message batches
Change-Id: I1d06486ccf7b174ee9aa621fa3d8acb8b3673777
Reviewed-on: https://code.wireshark.org/review/34222
Petri-Dish: Anders Broman <a.broman58@gmail.com>
Tested-by: Petri Dish Buildbot
Reviewed-by: Anders Broman <a.broman58@gmail.com>
Diffstat (limited to 'epan')
| -rw-r--r-- | epan/CMakeLists.txt | 2 | ||||
| -rw-r--r-- | epan/dissectors/packet-kafka.c | 42 | ||||
| -rw-r--r-- | epan/epan.c | 8 |
3 files changed, 52 insertions, 0 deletions
diff --git a/epan/CMakeLists.txt b/epan/CMakeLists.txt index 5bf71f4e2b..03f4939d2f 100644 --- a/epan/CMakeLists.txt +++ b/epan/CMakeLists.txt @@ -27,6 +27,7 @@ include_directories( ${SMI_INCLUDE_DIRS} ${SNAPPY_INCLUDE_DIRS} ${ZLIB_INCLUDE_DIRS} + ${ZSTD_INCLUDE_DIRS} ${LIBXML2_INCLUDE_DIRS} ) @@ -352,6 +353,7 @@ target_link_libraries(epan ${WIN_PSAPI_LIBRARY} ${LIBXML2_LIBRARIES} ${ZLIB_LIBRARIES} + ${ZSTD_LIBRARIES} ) target_include_directories(epan diff --git a/epan/dissectors/packet-kafka.c b/epan/dissectors/packet-kafka.c index 6fc5b37607..d3e2671bcf 100644 --- a/epan/dissectors/packet-kafka.c +++ b/epan/dissectors/packet-kafka.c @@ -26,6 +26,9 @@ #include <lz4.h> #include <lz4frame.h> #endif +#ifdef HAVE_ZSTD +#include <zstd.h> +#endif #include "packet-tcp.h" #include "packet-tls.h" @@ -1522,12 +1525,51 @@ decompress_snappy(tvbuff_t *tvb _U_, packet_info *pinfo, int offset _U_, int len } #endif /* HAVE_SNAPPY */ +#ifdef HAVE_ZSTD +static int +decompress_zstd(tvbuff_t *tvb, packet_info *pinfo, int offset, int length, tvbuff_t **decompressed_tvb, int *decompressed_offset) +{ + ZSTD_inBuffer input = { tvb_memdup(wmem_packet_scope(), tvb, offset, length), length, 0 }; + ZSTD_DStream *zds = ZSTD_createDStream(); + size_t rc = 0; + tvbuff_t *composite_tvb = tvb_new_composite(); + int ret = 0; + + do { + ZSTD_outBuffer output = { wmem_alloc(pinfo->pool, ZSTD_DStreamOutSize()), ZSTD_DStreamOutSize(), 0 }; + rc = ZSTD_decompressStream(zds, &output, &input); + // rc holds either the number of decompressed offsets or the error code. + // Both values are positive, one has to use ZSTD_isError to determine if the call succeeded. + if (ZSTD_isError(rc)) { + goto end; + } + tvb_composite_append(composite_tvb, + tvb_new_child_real_data(tvb, (guint8*)output.dst, (guint)output.pos, (gint)output.pos)); + // rc == 0 means there is nothing more to decompress, but there could be still something in the data + } while (rc > 0); + tvb_composite_finalize(composite_tvb); + *decompressed_tvb = composite_tvb; + *decompressed_offset = 0; + composite_tvb = NULL; + ret = 1; +end: + ZSTD_freeDStream(zds); + if (composite_tvb != NULL) { + tvb_free_chain(composite_tvb); + } + if (ret == 0) { + col_append_str(pinfo->cinfo, COL_INFO, " [zstd decompression failed]"); + } + return ret; +} +#else static int decompress_zstd(tvbuff_t *tvb _U_, packet_info *pinfo, int offset _U_, int length _U_, tvbuff_t **decompressed_tvb _U_, int *decompressed_offset _U_) { col_append_str(pinfo->cinfo, COL_INFO, " [zstd compression unsupported]"); return 0; } +#endif /* HAVE_ZSTD */ static int decompress(tvbuff_t *tvb, packet_info *pinfo, int offset, int length, int codec, tvbuff_t **decompressed_tvb, int *decompressed_offset) diff --git a/epan/epan.c b/epan/epan.c index ad51670088..cc4327d270 100644 --- a/epan/epan.c +++ b/epan/epan.c @@ -799,6 +799,14 @@ epan_get_compiled_version_info(GString *str) g_string_append(str, "without LZ4"); #endif /* HAVE_LZ4 */ + /* Zstandard */ + g_string_append(str, ", "); +#ifdef HAVE_ZSTD + g_string_append(str, "with Zstandard"); +#else + g_string_append(str, "without Zstandard"); +#endif /* HAVE_ZSTD */ + /* Snappy */ g_string_append(str, ", "); #ifdef HAVE_SNAPPY |