Struct EnvoyFilterInstance 

pub struct EnvoyFilterInstance { /* private fields */ }

An opaque object that represents the underlying Envoy Http filter instance. This is used to interact with it from the module code.

This is a shallow wrapper around the raw pointer to the Envoy filter instance. Can be copied and stored somewhere else. However, the object MUST NOT be used after the HttpFilterInstance::destroy for the corresponding filter instance is called.

Implementations

impl EnvoyFilterInstance

pub fn continue_request(&self)

Used to resume the request processing after the filter has stopped it.

Examples found in repository

example/example.rs (line 222)

    fn request_headers(
        &mut self,
        _request_headers: &RequestHeaders,
        _end_of_stream: bool,
    ) -> RequestHeadersStatus {
        if self.req_no == 1 {
            let envoy_filter_instance = self.envoy_filter_instance.clone();
            let req_no = self.req_no;
            std::thread::spawn(move || {
                println!("blocking for 1 second at RequestHeaders with id {}", req_no);
                std::thread::sleep(std::time::Duration::from_secs(1));
                println!("calling ContinueRequest with id {}", req_no);
                if let Some(envoy_filter_instance) = envoy_filter_instance.lock().unwrap().as_ref()
                {
                    envoy_filter_instance.continue_request();
                }
            });
            println!(
                "RequestHeaders returning StopAllIterationAndBuffer with id {}",
                self.req_no
            );
            RequestHeadersStatus::StopAllIterationAndBuffer
        } else {
            println!("RequestHeaders called with id {}", self.req_no);
            RequestHeadersStatus::Continue
        }
    }

    fn request_body(
        &mut self,
        _request_body_frame: &RequestBodyBuffer,
        _end_of_stream: bool,
    ) -> RequestBodyStatus {
        if self.req_no == 2 {
            let envoy_filter_instance = self.envoy_filter_instance.clone();
            let req_no = self.req_no;
            std::thread::spawn(move || {
                println!("blocking for 1 second at RequestBody with id {}", req_no);
                std::thread::sleep(std::time::Duration::from_secs(1));
                println!("calling ContinueRequest with id {}", req_no);
                if let Some(envoy_filter_instance) = envoy_filter_instance.lock().unwrap().as_ref()
                {
                    envoy_filter_instance.continue_request();
                }
            });
            println!(
                "RequestBody returning StopIterationAndBuffer with id {}",
                self.req_no
            );
            RequestBodyStatus::StopIterationAndBuffer
        } else {
            println!("RequestBody called with id {}", self.req_no);
            RequestBodyStatus::Continue
        }
    }

pub fn continue_response(&self)

Used to resume the response processing after the filter has stopped it.

Examples found in repository

example/example.rs (line 281)

    fn response_headers(
        &mut self,
        _response_headers: &ResponseHeaders,
        _end_of_stream: bool,
    ) -> ResponseHeadersStatus {
        if self.req_no == 3 {
            let envoy_filter_instance = self.envoy_filter_instance.clone();
            let req_no = self.req_no;
            std::thread::spawn(move || {
                println!(
                    "blocking for 1 second at ResponseHeaders with id {}",
                    req_no
                );
                std::thread::sleep(std::time::Duration::from_secs(1));
                println!("calling ContinueResponse with id {}", req_no);
                if let Some(envoy_filter_instance) = envoy_filter_instance.lock().unwrap().as_ref()
                {
                    envoy_filter_instance.continue_response();
                }
            });
            println!(
                "ResponseHeaders returning StopAllIterationAndBuffer with id {}",
                self.req_no
            );

            ResponseHeadersStatus::StopAllIterationAndBuffer
        } else {
            println!("ResponseHeaders called with id {}", self.req_no);
            ResponseHeadersStatus::Continue
        }
    }

    fn response_body(
        &mut self,
        _response_body_frame: &ResponseBodyBuffer,
        _end_of_stream: bool,
    ) -> ResponseBodyStatus {
        if self.req_no == 4 {
            let envoy_filter_instance = self.envoy_filter_instance.clone();
            let req_no = self.req_no;
            std::thread::spawn(move || {
                println!("blocking for 1 second at ResponseBody with id {}", req_no);
                std::thread::sleep(std::time::Duration::from_secs(1));
                println!("calling ContinueResponse with id {}", req_no);
                if let Some(envoy_filter_instance) = envoy_filter_instance.lock().unwrap().as_ref()
                {
                    envoy_filter_instance.continue_response();
                }
            });
            println!(
                "ResponseBody returning StopIterationAndBuffer with id {}",
                self.req_no
            );

            ResponseBodyStatus::StopIterationAndBuffer
        } else {
            println!("ResponseBody called with id {}", self.req_no);
            ResponseBodyStatus::Continue
        }
    }

pub fn get_request_body_buffer(&self) -> RequestBodyBuffer

Returns the entire request body buffer.

Examples found in repository

example/example.rs (line 369)

    fn request_body(
        &mut self,
        request_body_frame: &RequestBodyBuffer,
        end_of_stream: bool,
    ) -> RequestBodyStatus {
        println!(
            "new request body frame: {}",
            String::from_utf8(request_body_frame.copy()).unwrap()
        );
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return RequestBodyStatus::StopIterationAndBuffer;
        }

        // Get the entire request body reference - this does not copy the body.
        let entire_body = self.envoy_filter_instance.get_request_body_buffer();
        println!(
            "entire request body: {}",
            String::from_utf8(entire_body.copy()).unwrap()
        );

        // This demonstrates how to use the reader to read the body.
        let mut reader = entire_body.reader();
        let mut buf = vec![0; 2];
        let mut offset = 0;
        loop {
            let n = reader.read(&mut buf).unwrap();
            if n == 0 {
                break;
            }
            println!(
                "request body read 2 bytes offset at {}: \"{}\"",
                offset,
                std::str::from_utf8(&buf[..n]).unwrap()
            );
            offset += 2;
        }

        // Replace the entire body with 'Y' without copying.
        for i in entire_body.slices() {
            for j in i {
                *j = b'X';
            }
        }
        RequestBodyStatus::Continue
    }

    fn response_body(
        &mut self,
        response_body_frame: &ResponseBodyBuffer,
        end_of_stream: bool,
    ) -> ResponseBodyStatus {
        println!(
            "new response body frame: {}",
            String::from_utf8(response_body_frame.copy()).unwrap()
        );
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return ResponseBodyStatus::StopIterationAndBuffer;
        }

        // Get the entire response body reference - this does not copy the body.
        let entire_body = self.envoy_filter_instance.get_response_body_buffer();
        println!(
            "entire response body: {}",
            String::from_utf8(entire_body.copy()).unwrap()
        );

        // This demonstrates how to use the reader to read the body.
        let mut reader = entire_body.reader();
        let mut buf = vec![0; 2];
        let mut offset = 0;
        loop {
            let n = reader.read(&mut buf).unwrap();
            if n == 0 {
                break;
            }
            println!(
                "response body read 2 bytes offset at {}: \"{}\"",
                offset,
                std::str::from_utf8(&buf[..n]).unwrap()
            );
            offset += 2;
        }

        // Replace the entire body with 'Y' without copying.
        for i in entire_body.slices() {
            for j in i {
                *j = b'Y';
            }
        }

        ResponseBodyStatus::Continue
    }
}

/// BodiesReplaceFilter is a filter that replaces request/response bodies.
///
/// This implements the [`HttpFilter`] trait, and will be created per each filter chain.
struct BodiesReplace {}

impl HttpFilter for BodiesReplace {
    fn new_instance(
        &mut self,
        envoy_filter_instance: EnvoyFilterInstance,
    ) -> Box<dyn HttpFilterInstance> {
        Box::new(BodiesReplaceInstance {
            envoy_filter_instance,
            request_append: String::new(),
            request_prepend: String::new(),
            request_replace: String::new(),
            response_append: String::new(),
            response_prepend: String::new(),
            response_replace: String::new(),
        })
    }
}

/// BodiesReplaceInstance is a filter instance that replaces request/response bodies.
///
/// This implements the [`HttpFilterInstance`] trait, and will be created per each request.
struct BodiesReplaceInstance {
    envoy_filter_instance: EnvoyFilterInstance,
    request_append: String,
    request_prepend: String,
    request_replace: String,
    response_append: String,
    response_prepend: String,
    response_replace: String,
}

impl HttpFilterInstance for BodiesReplaceInstance {
    fn request_headers(
        &mut self,
        request_headers: &RequestHeaders,
        _end_of_stream: bool,
    ) -> RequestHeadersStatus {
        if let Some(value) = request_headers.get(b"append") {
            self.request_append = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = request_headers.get(b"prepend") {
            self.request_prepend = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = request_headers.get(b"replace") {
            self.request_replace = std::str::from_utf8(value).unwrap().to_string();
        }
        request_headers.remove(b"content-length"); // Remove content-length header to avoid mismatch.
        RequestHeadersStatus::Continue
    }

    fn request_body(
        &mut self,
        _request_body_frame: &RequestBodyBuffer,
        end_of_stream: bool,
    ) -> RequestBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return RequestBodyStatus::StopIterationAndBuffer;
        }

        let entire_body = self.envoy_filter_instance.get_request_body_buffer();
        if !self.request_append.is_empty() {
            entire_body.append(self.request_append.as_bytes());
        }
        if !self.request_prepend.is_empty() {
            entire_body.prepend(self.request_prepend.as_bytes());
        }
        if !self.request_replace.is_empty() {
            entire_body.replace(self.request_replace.as_bytes());
        }
        RequestBodyStatus::Continue
    }

    fn response_headers(
        &mut self,
        response_headers: &ResponseHeaders,
        _end_of_stream: bool,
    ) -> ResponseHeadersStatus {
        if let Some(value) = response_headers.get(b"append") {
            self.response_append = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = response_headers.get(b"prepend") {
            self.response_prepend = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = response_headers.get(b"replace") {
            self.response_replace = std::str::from_utf8(value).unwrap().to_string();
        }
        response_headers.remove(b"content-length"); // Remove content-length header to avoid mismatch.
        ResponseHeadersStatus::Continue
    }

    fn response_body(
        &mut self,
        _response_body_frame: &ResponseBodyBuffer,
        end_of_stream: bool,
    ) -> ResponseBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return ResponseBodyStatus::StopIterationAndBuffer;
        }

        let entire_body = self.envoy_filter_instance.get_response_body_buffer();
        if !self.response_append.is_empty() {
            entire_body.append(self.response_append.as_bytes());
        }
        if !self.response_prepend.is_empty() {
            entire_body.prepend(self.response_prepend.as_bytes());
        }
        if !self.response_replace.is_empty() {
            entire_body.replace(self.response_replace.as_bytes());
        }
        ResponseBodyStatus::Continue
    }
}

/// SendResponseFilter is a filter that sends a response.
///
/// This implements the [`HttpFilter`] trait, and will be created per each filter chain.
struct SendResponseFilter {}

impl HttpFilter for SendResponseFilter {
    fn new_instance(
        &mut self,
        envoy_filter_instance: EnvoyFilterInstance,
    ) -> Box<dyn HttpFilterInstance> {
        Box::new(SendResponseFilterInstance {
            envoy_filter_instance,
            on_response_headers: false,
        })
    }
}

/// SendResponseFilterInstance is a filter instance that sends a response.
///
/// This implements the [`HttpFilterInstance`] trait, and will be created per each request.
struct SendResponseFilterInstance {
    envoy_filter_instance: EnvoyFilterInstance,
    on_response_headers: bool,
}

impl HttpFilterInstance for SendResponseFilterInstance {
    fn request_headers(
        &mut self,
        request_headers: &RequestHeaders,
        _end_of_stream: bool,
    ) -> RequestHeadersStatus {
        if let Some(value) = request_headers.get(b":path") {
            if value == "/on_request".as_bytes() {
                let headers: Vec<(&[u8], &[u8])> = vec![
                    ("foo".as_bytes(), "bar".as_bytes()),
                    ("bar".as_bytes(), "baz".as_bytes()),
                ];
                self.envoy_filter_instance.send_response(
                    401,
                    headers.as_slice(),
                    "local response at request headers".as_bytes(),
                );
            }
            if value == b"/on_response" {
                self.on_response_headers = true;
            }
        }
        RequestHeadersStatus::Continue
    }

    fn response_headers(
        &mut self,
        _response_headers: &ResponseHeaders,
        _end_of_stream: bool,
    ) -> ResponseHeadersStatus {
        if self.on_response_headers {
            let headers: Vec<(&[u8], &[u8])> = vec![("dog".as_bytes(), "cat".as_bytes())];
            self.envoy_filter_instance.send_response(
                500,
                headers.as_slice(),
                "local response at response headers".as_bytes(),
            );
        }
        ResponseHeadersStatus::Continue
    }
}

/// ValidateJsonFilter is a filter that validates JSON.
///
/// This implements the [`HttpFilter`] trait, and will be created per each filter chain.
struct ValidateJsonFilter {}

impl HttpFilter for ValidateJsonFilter {
    fn new_instance(
        &mut self,
        envoy_filter_instance: EnvoyFilterInstance,
    ) -> Box<dyn HttpFilterInstance> {
        Box::new(ValidateJsonFilterInstance {
            envoy_filter_instance,
        })
    }
}

/// ValidateJsonFilterInstance is a filter instance that validates JSON.
///
/// This implements the [`HttpFilterInstance`] trait, and will be created per each request.
struct ValidateJsonFilterInstance {
    envoy_filter_instance: EnvoyFilterInstance,
}

#[derive(Serialize, Deserialize)]
struct ValidateJsonFilterBody {
    foo: String,
}

impl HttpFilterInstance for ValidateJsonFilterInstance {
    fn request_body(
        &mut self,
        _request_body_frame: &RequestBodyBuffer,
        end_of_stream: bool,
    ) -> RequestBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return RequestBodyStatus::StopIterationAndBuffer;
        }

        let reader = self
            .envoy_filter_instance
            .get_request_body_buffer()
            .reader();

        match serde_json::from_reader(reader) {
            Ok(body) => {
                let _body: ValidateJsonFilterBody = body;
            }
            Err(_e) => {
                self.envoy_filter_instance.send_response(400, &[], &[]);
            }
        }
        RequestBodyStatus::Continue
    }

pub fn get_response_body_buffer(&self) -> ResponseBodyBuffer

Returns the entire request body buffer.

Examples found in repository

example/example.rs (line 416)

    fn response_body(
        &mut self,
        response_body_frame: &ResponseBodyBuffer,
        end_of_stream: bool,
    ) -> ResponseBodyStatus {
        println!(
            "new response body frame: {}",
            String::from_utf8(response_body_frame.copy()).unwrap()
        );
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return ResponseBodyStatus::StopIterationAndBuffer;
        }

        // Get the entire response body reference - this does not copy the body.
        let entire_body = self.envoy_filter_instance.get_response_body_buffer();
        println!(
            "entire response body: {}",
            String::from_utf8(entire_body.copy()).unwrap()
        );

        // This demonstrates how to use the reader to read the body.
        let mut reader = entire_body.reader();
        let mut buf = vec![0; 2];
        let mut offset = 0;
        loop {
            let n = reader.read(&mut buf).unwrap();
            if n == 0 {
                break;
            }
            println!(
                "response body read 2 bytes offset at {}: \"{}\"",
                offset,
                std::str::from_utf8(&buf[..n]).unwrap()
            );
            offset += 2;
        }

        // Replace the entire body with 'Y' without copying.
        for i in entire_body.slices() {
            for j in i {
                *j = b'Y';
            }
        }

        ResponseBodyStatus::Continue
    }
}

/// BodiesReplaceFilter is a filter that replaces request/response bodies.
///
/// This implements the [`HttpFilter`] trait, and will be created per each filter chain.
struct BodiesReplace {}

impl HttpFilter for BodiesReplace {
    fn new_instance(
        &mut self,
        envoy_filter_instance: EnvoyFilterInstance,
    ) -> Box<dyn HttpFilterInstance> {
        Box::new(BodiesReplaceInstance {
            envoy_filter_instance,
            request_append: String::new(),
            request_prepend: String::new(),
            request_replace: String::new(),
            response_append: String::new(),
            response_prepend: String::new(),
            response_replace: String::new(),
        })
    }
}

/// BodiesReplaceInstance is a filter instance that replaces request/response bodies.
///
/// This implements the [`HttpFilterInstance`] trait, and will be created per each request.
struct BodiesReplaceInstance {
    envoy_filter_instance: EnvoyFilterInstance,
    request_append: String,
    request_prepend: String,
    request_replace: String,
    response_append: String,
    response_prepend: String,
    response_replace: String,
}

impl HttpFilterInstance for BodiesReplaceInstance {
    fn request_headers(
        &mut self,
        request_headers: &RequestHeaders,
        _end_of_stream: bool,
    ) -> RequestHeadersStatus {
        if let Some(value) = request_headers.get(b"append") {
            self.request_append = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = request_headers.get(b"prepend") {
            self.request_prepend = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = request_headers.get(b"replace") {
            self.request_replace = std::str::from_utf8(value).unwrap().to_string();
        }
        request_headers.remove(b"content-length"); // Remove content-length header to avoid mismatch.
        RequestHeadersStatus::Continue
    }

    fn request_body(
        &mut self,
        _request_body_frame: &RequestBodyBuffer,
        end_of_stream: bool,
    ) -> RequestBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return RequestBodyStatus::StopIterationAndBuffer;
        }

        let entire_body = self.envoy_filter_instance.get_request_body_buffer();
        if !self.request_append.is_empty() {
            entire_body.append(self.request_append.as_bytes());
        }
        if !self.request_prepend.is_empty() {
            entire_body.prepend(self.request_prepend.as_bytes());
        }
        if !self.request_replace.is_empty() {
            entire_body.replace(self.request_replace.as_bytes());
        }
        RequestBodyStatus::Continue
    }

    fn response_headers(
        &mut self,
        response_headers: &ResponseHeaders,
        _end_of_stream: bool,
    ) -> ResponseHeadersStatus {
        if let Some(value) = response_headers.get(b"append") {
            self.response_append = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = response_headers.get(b"prepend") {
            self.response_prepend = std::str::from_utf8(value).unwrap().to_string();
        }
        if let Some(value) = response_headers.get(b"replace") {
            self.response_replace = std::str::from_utf8(value).unwrap().to_string();
        }
        response_headers.remove(b"content-length"); // Remove content-length header to avoid mismatch.
        ResponseHeadersStatus::Continue
    }

    fn response_body(
        &mut self,
        _response_body_frame: &ResponseBodyBuffer,
        end_of_stream: bool,
    ) -> ResponseBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return ResponseBodyStatus::StopIterationAndBuffer;
        }

        let entire_body = self.envoy_filter_instance.get_response_body_buffer();
        if !self.response_append.is_empty() {
            entire_body.append(self.response_append.as_bytes());
        }
        if !self.response_prepend.is_empty() {
            entire_body.prepend(self.response_prepend.as_bytes());
        }
        if !self.response_replace.is_empty() {
            entire_body.replace(self.response_replace.as_bytes());
        }
        ResponseBodyStatus::Continue
    }

pub fn send_response( &self, status_code: u32, headers: &[(&[u8], &[u8])], body: &[u8], )

Sends the response to the downstream.

• status_code is the HTTP status code.
• headers is the list of headers. Each header is a tuple of key and value.
• body is the response body.

Examples found in repository

example/example.rs (lines 606-610)

    fn request_headers(
        &mut self,
        request_headers: &RequestHeaders,
        _end_of_stream: bool,
    ) -> RequestHeadersStatus {
        if let Some(value) = request_headers.get(b":path") {
            if value == "/on_request".as_bytes() {
                let headers: Vec<(&[u8], &[u8])> = vec![
                    ("foo".as_bytes(), "bar".as_bytes()),
                    ("bar".as_bytes(), "baz".as_bytes()),
                ];
                self.envoy_filter_instance.send_response(
                    401,
                    headers.as_slice(),
                    "local response at request headers".as_bytes(),
                );
            }
            if value == b"/on_response" {
                self.on_response_headers = true;
            }
        }
        RequestHeadersStatus::Continue
    }

    fn response_headers(
        &mut self,
        _response_headers: &ResponseHeaders,
        _end_of_stream: bool,
    ) -> ResponseHeadersStatus {
        if self.on_response_headers {
            let headers: Vec<(&[u8], &[u8])> = vec![("dog".as_bytes(), "cat".as_bytes())];
            self.envoy_filter_instance.send_response(
                500,
                headers.as_slice(),
                "local response at response headers".as_bytes(),
            );
        }
        ResponseHeadersStatus::Continue
    }
}

/// ValidateJsonFilter is a filter that validates JSON.
///
/// This implements the [`HttpFilter`] trait, and will be created per each filter chain.
struct ValidateJsonFilter {}

impl HttpFilter for ValidateJsonFilter {
    fn new_instance(
        &mut self,
        envoy_filter_instance: EnvoyFilterInstance,
    ) -> Box<dyn HttpFilterInstance> {
        Box::new(ValidateJsonFilterInstance {
            envoy_filter_instance,
        })
    }
}

/// ValidateJsonFilterInstance is a filter instance that validates JSON.
///
/// This implements the [`HttpFilterInstance`] trait, and will be created per each request.
struct ValidateJsonFilterInstance {
    envoy_filter_instance: EnvoyFilterInstance,
}

#[derive(Serialize, Deserialize)]
struct ValidateJsonFilterBody {
    foo: String,
}

impl HttpFilterInstance for ValidateJsonFilterInstance {
    fn request_body(
        &mut self,
        _request_body_frame: &RequestBodyBuffer,
        end_of_stream: bool,
    ) -> RequestBodyStatus {
        if !end_of_stream {
            // Wait for the end of the stream to see the full body.
            return RequestBodyStatus::StopIterationAndBuffer;
        }

        let reader = self
            .envoy_filter_instance
            .get_request_body_buffer()
            .reader();

        match serde_json::from_reader(reader) {
            Ok(body) => {
                let _body: ValidateJsonFilterBody = body;
            }
            Err(_e) => {
                self.envoy_filter_instance.send_response(400, &[], &[]);
            }
        }
        RequestBodyStatus::Continue
    }

Trait Implementations

impl Clone for EnvoyFilterInstance

fn clone(&self) -> EnvoyFilterInstance

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for EnvoyFilterInstance

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Copy for EnvoyFilterInstance