|
| 1 | +# frozen_string_literal: true |
| 2 | + |
| 3 | +require 'stringio' |
| 4 | +require 'datadog/core/utils/base64' |
| 5 | + |
| 6 | +module Datadog |
| 7 | + module DataStreams |
| 8 | + # Represents a pathway context for data streams monitoring |
| 9 | + class PathwayContext |
| 10 | + attr_accessor :hash, |
| 11 | + :pathway_start_sec, |
| 12 | + :current_edge_start_sec, |
| 13 | + :parent_hash, |
| 14 | + :previous_direction, |
| 15 | + :closest_opposite_direction_hash, |
| 16 | + :closest_opposite_direction_edge_start |
| 17 | + |
| 18 | + def initialize(hash_value:, pathway_start_sec:, current_edge_start_sec:) |
| 19 | + @hash = hash_value |
| 20 | + @pathway_start_sec = pathway_start_sec |
| 21 | + @current_edge_start_sec = current_edge_start_sec |
| 22 | + @parent_hash = nil |
| 23 | + |
| 24 | + @previous_direction = '' |
| 25 | + @closest_opposite_direction_hash = 0 |
| 26 | + @closest_opposite_direction_edge_start = current_edge_start_sec |
| 27 | + end |
| 28 | + |
| 29 | + def encode |
| 30 | + # Format: |
| 31 | + # - 8 bytes: hash value (little-endian) |
| 32 | + # - VarInt: pathway start time (milliseconds) |
| 33 | + # - VarInt: current edge start time (milliseconds) |
| 34 | + [@hash].pack('Q') << |
| 35 | + encode_var_int_64((@pathway_start_sec * 1000).to_i) << |
| 36 | + encode_var_int_64((@current_edge_start_sec * 1000).to_i) |
| 37 | + end |
| 38 | + |
| 39 | + def encode_b64 |
| 40 | + Core::Utils::Base64.strict_encode64(encode) |
| 41 | + end |
| 42 | + |
| 43 | + # Decode pathway context from base64 encoded string |
| 44 | + def self.decode_b64(encoded_ctx) |
| 45 | + return nil unless encoded_ctx && !encoded_ctx.empty? |
| 46 | + |
| 47 | + begin |
| 48 | + binary_data = Core::Utils::Base64.strict_decode64(encoded_ctx) |
| 49 | + decode(binary_data) |
| 50 | + rescue |
| 51 | + # Invalid base64 or decode error |
| 52 | + nil |
| 53 | + end |
| 54 | + end |
| 55 | + |
| 56 | + # Decode pathway context from binary data |
| 57 | + def self.decode(binary_data) |
| 58 | + return nil unless binary_data && binary_data.bytesize >= 8 |
| 59 | + |
| 60 | + reader = StringIO.new(binary_data) |
| 61 | + |
| 62 | + # Extract 8-byte hash (little-endian) |
| 63 | + hash_bytes = reader.read(8) |
| 64 | + return nil unless hash_bytes |
| 65 | + |
| 66 | + hash_value = hash_bytes.unpack1('Q') |
| 67 | + |
| 68 | + # Extract pathway start time (VarInt milliseconds) |
| 69 | + pathway_start_ms = decode_varint(reader) |
| 70 | + return nil unless pathway_start_ms |
| 71 | + |
| 72 | + # Extract current edge start time (VarInt milliseconds) |
| 73 | + current_edge_start_ms = decode_varint(reader) |
| 74 | + return nil unless current_edge_start_ms |
| 75 | + |
| 76 | + # Convert milliseconds to seconds |
| 77 | + pathway_start_sec = pathway_start_ms / 1000.0 |
| 78 | + current_edge_start_sec = current_edge_start_ms / 1000.0 |
| 79 | + |
| 80 | + new( |
| 81 | + hash_value: hash_value, |
| 82 | + pathway_start_sec: pathway_start_sec, |
| 83 | + current_edge_start_sec: current_edge_start_sec |
| 84 | + ) |
| 85 | + rescue EOFError |
| 86 | + # Not enough data in binary stream |
| 87 | + nil |
| 88 | + end |
| 89 | + |
| 90 | + private |
| 91 | + |
| 92 | + def encode_var_int_64(value) |
| 93 | + bytes = [] |
| 94 | + while value >= 0x80 |
| 95 | + bytes << ((value & 0x7F) | 0x80) |
| 96 | + value >>= 7 |
| 97 | + end |
| 98 | + bytes << value |
| 99 | + bytes.pack('C*') |
| 100 | + end |
| 101 | + |
| 102 | + # Decode VarInt from IO stream using Ruby-idiomatic approach |
| 103 | + # |
| 104 | + # VarInt format: Each byte uses 7 bits for data, 1 bit for continuation |
| 105 | + # - High bit set = more bytes follow |
| 106 | + # - High bit clear = final byte |
| 107 | + # - Data bits accumulated in little-endian order |
| 108 | + def self.decode_varint(io) |
| 109 | + value = 0 |
| 110 | + shift = 0 |
| 111 | + |
| 112 | + loop do |
| 113 | + byte = io.readbyte |
| 114 | + |
| 115 | + # Add this byte's 7 data bits to our value |
| 116 | + value |= (byte & 0x7F) << shift |
| 117 | + |
| 118 | + # If high bit is clear, we're done |
| 119 | + return value unless (byte & 0x80).nonzero? |
| 120 | + |
| 121 | + shift += 7 |
| 122 | + |
| 123 | + # Safety: prevent infinite decoding |
| 124 | + raise EOFError if shift >= 64 |
| 125 | + end |
| 126 | + rescue EOFError |
| 127 | + nil |
| 128 | + end |
| 129 | + private_class_method :decode_varint |
| 130 | + end |
| 131 | + end |
| 132 | +end |
0 commit comments