Method on_read

rust/src/network_redis.rs:336–402

(
        &mut self,
        envoy_filter: &mut ENF,
        read_buffer_length: usize,
        _end_stream: bool,
    )

    }

    fn on_read(
        &mut self,
        envoy_filter: &mut ENF,
        read_buffer_length: usize,
        _end_stream: bool,
    ) -> abi::envoy_dynamic_module_type_on_network_filter_data_status {
        let _ = envoy_filter.increment_counter(self.bytes_received, read_buffer_length as u64);

        // Get data from read buffer.
        let (chunks, _) = envoy_filter.get_read_buffer_chunks();

        // Collect all bytes.
        let mut data = Vec::new();
        for chunk in chunks {
            data.extend_from_slice(chunk.as_slice());
        }

        // Check for command length limit.
        if data.len() > self.max_command_length {
            envoy_log_info!(
                "Redis command exceeds max length: {} > {}",
                data.len(),
                self.max_command_length
            );
            let _ = envoy_filter.increment_counter(self.parse_errors, 1);
            envoy_filter.drain_read_buffer(read_buffer_length);
            envoy_filter.write(&self.create_error_response("command too long"), false);
            return abi::envoy_dynamic_module_type_on_network_filter_data_status::StopIteration;
        }

        // Parse Redis commands.
        match self.parse_commands(&data) {
            Ok(commands) => {
                for cmd in &commands {
                    let _ = envoy_filter.increment_counter(self.commands_total, 1);

                    if self.log_commands {
                        let args_str: Vec<String> = cmd
                            .args
                            .iter()
                            .take(3)
                            .map(|a| String::from_utf8_lossy(a).to_string())
                            .collect();
                        envoy_log_debug!("Redis command: {} {:?}", cmd.name, args_str);
                    }

                    // Check if command is blocked.
                    if self.blocked_commands.contains(&cmd.name) {
                        envoy_log_info!("Blocked Redis command: {}", cmd.name);
                        let _ = envoy_filter.increment_counter(self.commands_blocked, 1);

                        // Drain the read buffer and send error response.
                        envoy_filter.drain_read_buffer(read_buffer_length);
                        let error_msg = format!("command '{}' is not allowed", cmd.name);
                        envoy_filter.write(&self.create_error_response(&error_msg), false);
                        return abi::envoy_dynamic_module_type_on_network_filter_data_status::StopIteration;
                    }
                }