1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
use crate::abi;
use crate::Error;
use fastly_shared::BodyWriteEnd;
use lazy_static::lazy_static;
use std::convert::TryFrom;
use std::io::{BufReader, BufWriter, Read, Write};
use std::sync::Mutex;

/// A low-level interface to HTTP bodies.
///
/// This type implements `Read` to read bytes from the beginning of a body, and `Write` to write to
/// the end of a body. Note that these operations are unbuffered, unlike the same operations on the
/// higher-level `Body` type.
///
/// These handles can also be used to append bodies to each other in amortized constant time.
#[derive(Debug, Eq, Hash, PartialEq)]
#[repr(transparent)]
pub struct BodyHandle {
    handle: u32,
}

lazy_static! {
    pub(crate) static ref GOT_DOWNSTREAM: Mutex<bool> = Mutex::new(false);
}

impl BodyHandle {
    pub const INVALID: Self = Self {
        handle: fastly_shared::INVALID_BODY_HANDLE,
    };

    pub fn is_invalid(&self) -> bool {
        self.handle == Self::INVALID.handle
    }

    /// Get an owned `BodyHandle` from a borrowed one.
    ///
    /// This should only be used when calling the raw ABI directly.
    pub(crate) unsafe fn handle(&self) -> Self {
        Self {
            handle: self.handle,
        }
    }

    /// Get a handle to the downstream request body.
    ///
    /// This handle may only be retrieved once per execution, either through this function or
    /// through `fastly::request::downstream_request_and_body_handles()`.
    pub fn downstream_request() -> Result<Self, Error> {
        let mut got = GOT_DOWNSTREAM.lock().unwrap();
        if *got {
            return Err(Error::new(
                "cannot get more than one handle to the downstream request body per execution",
            ));
        }

        let mut handle = BodyHandle::INVALID;
        let status = unsafe { abi::xqd_req_body_downstream_get(std::ptr::null_mut(), &mut handle) };
        if status.is_err() || handle.is_invalid() {
            Err(Error::new("xqd_req_body_downstream_get failed"))
        } else {
            *got = true;
            Ok(handle)
        }
    }

    /// Get a handle to a new, empty body.
    pub fn new() -> Result<BodyHandle, Error> {
        let mut handle = BodyHandle::INVALID;
        let status = unsafe { abi::xqd_body_new(&mut handle) };
        if status.is_err() || handle.is_invalid() {
            Err(Error::new("xqd_body_new failed"))
        } else {
            Ok(handle)
        }
    }

    /// Append another body onto the end of this body.
    ///
    /// The other body will no longer be valid after this call.
    pub fn append(&mut self, other: BodyHandle) -> Result<(), Error> {
        let status = unsafe { abi::xqd_body_append(self.handle(), other.handle()) };
        if status.is_err() {
            Err(Error::new("xqd_body_append failed"))
        } else {
            Ok(())
        }
    }

    /// Read the entirety of the body.
    ///
    /// Note that this will involve copying and buffering the body, and so should only be used for
    /// convenience on small request bodies.
    pub fn into_bytes(self) -> Result<Vec<u8>, Error> {
        let mut body = vec![];
        let mut bufread = BufReader::new(self);
        bufread
            .read_to_end(&mut body)
            .map_err(|e| Error::new(format!("{}", e)))?;
        Ok(body)
    }

    /// Read the entirety of the body into a `String`, interpreting the bytes as UTF-8.
    ///
    /// Note that this will involve copying and buffering the body, and so should only be used for
    /// convenience on small request bodies.
    pub fn into_string(self) -> Result<String, Error> {
        let mut body = String::new();
        let mut bufread = BufReader::new(self);
        bufread
            .read_to_string(&mut body)
            .map_err(|e| Error::new(format!("{}", e)))?;
        Ok(body)
    }
}

impl Read for BodyHandle {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        use std::io::{Error, ErrorKind};

        let mut nread = 0;
        let status =
            unsafe { abi::xqd_body_read(self.handle(), buf.as_mut_ptr(), buf.len(), &mut nread) };
        if status.is_err() {
            Err(Error::new(ErrorKind::Other, "xqd_body_read failed"))
        } else {
            Ok(nread)
        }
    }
}

impl Write for BodyHandle {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        use std::io::{Error, ErrorKind};

        let mut nwritten = 0;
        let status = unsafe {
            abi::xqd_body_write(
                self.handle(),
                buf.as_ptr(),
                buf.len(),
                BodyWriteEnd::Back,
                &mut nwritten,
            )
        };
        if status.is_err() {
            Err(Error::new(ErrorKind::Other, "xqd_body_write failed"))
        } else {
            Ok(nwritten)
        }
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

impl TryFrom<&str> for BodyHandle {
    type Error = Error;

    fn try_from(s: &str) -> Result<Self, Error> {
        let mut handle = BodyHandle::new()?;
        handle
            .write_all(s.as_bytes())
            .map_err(|e| Error::new(format!("{}", e)))?;
        Ok(handle)
    }
}

impl TryFrom<String> for BodyHandle {
    type Error = Error;

    fn try_from(s: String) -> Result<Self, Error> {
        BodyHandle::try_from(s.as_str())
    }
}

impl TryFrom<&[u8]> for BodyHandle {
    type Error = Error;

    fn try_from(s: &[u8]) -> Result<Self, Error> {
        let mut handle = BodyHandle::new()?;
        handle
            .write_all(s)
            .map_err(|e| Error::new(format!("{}", e)))?;
        Ok(handle)
    }
}

impl TryFrom<Vec<u8>> for BodyHandle {
    type Error = Error;

    fn try_from(s: Vec<u8>) -> Result<Self, Error> {
        BodyHandle::try_from(s.as_slice())
    }
}

/// An HTTP body that can be read from, written to, or appended to another body.
///
/// The most efficient way to read and write the body is through the `Read`, `BufRead`, and `Write`
/// implementations.
pub struct Body {
    reader: BufReader<BodyHandle>,
    writer: BufWriter<BodyHandle>,
}

impl Body {
    pub fn new() -> Result<Self, Error> {
        Ok(BodyHandle::new()?.into())
    }

    // this is not exported, since misuse can lead to data getting dropped or appearing out of order
    fn handle(&mut self) -> &mut BodyHandle {
        self.reader.get_mut()
    }

    pub fn into_handle(mut self) -> Result<BodyHandle, Error> {
        // we need to make sure that any currently buffered data in either the reader or writer is
        // put (back) in the host body
        let read_buf = self.reader.buffer();
        if !read_buf.is_empty() {
            let mut nwritten = 0;
            let status = unsafe {
                abi::xqd_body_write(
                    self.reader.get_ref().handle(),
                    read_buf.as_ptr(),
                    read_buf.len(),
                    BodyWriteEnd::Front,
                    &mut nwritten,
                )
            };
            if status.is_err() {
                return Err(Error::new("xqd_body_write_front failed"));
            }
            if nwritten != read_buf.len() {
                return Err(Error::new("xqd_body_write_front didn't fully write"));
            }
        }
        self.writer
            .flush()
            .map_err(|e| Error::new(format!("{}", e)))?;
        Ok(self.reader.into_inner())
    }

    /// Read the entirety of the body.
    ///
    /// Note that this will involve copying and buffering the body, and so should only be used for
    /// convenience on small request bodies.
    pub fn into_bytes(self) -> Result<Vec<u8>, Error> {
        self.into_handle()?.into_bytes()
    }

    /// Read the entirety of the body into a `String`, interpreting the bytes as UTF-8.
    ///
    /// Note that this will involve copying and buffering the body, and so should only be used for
    /// convenience on small request bodies.
    pub fn into_string(self) -> Result<String, Error> {
        self.into_handle()?.into_string()
    }

    /// Append another body onto the end of this body.
    pub fn append(&mut self, other: Body) -> Result<(), Error> {
        // flush the write buffer of the destination body, so that we can use the append method on
        // the underlying handles
        self.writer
            .flush()
            .map_err(|e| Error::new(format!("{}", e)))?;
        self.handle().append(other.into_handle()?)
    }
}

// For these trait implementations we only implement the methods that the underlying buffered
// adaptors implement; the default implementations for the others will behave the same.
//
// The main bit of caution we must use here is that any read should be preceded by flushing the
// write buffer. `BufWriter` doesn't make any calls if its buffer is empty, so this isn't very
// expensive and could prevent unexpected results if a program is trying to read and write from the
// same body.
impl Read for Body {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.writer.flush()?;
        self.reader.read(buf)
    }

    fn read_vectored(&mut self, bufs: &mut [std::io::IoSliceMut]) -> std::io::Result<usize> {
        self.writer.flush()?;
        self.reader.read_vectored(bufs)
    }
}

impl std::io::BufRead for Body {
    fn fill_buf(&mut self) -> std::io::Result<&[u8]> {
        self.writer.flush()?;
        self.reader.fill_buf()
    }

    fn consume(&mut self, amt: usize) {
        self.reader.consume(amt)
    }
}

impl Write for Body {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        self.writer.write(buf)
    }

    fn write_vectored(&mut self, bufs: &[std::io::IoSlice<'_>]) -> std::io::Result<usize> {
        self.writer.write_vectored(bufs)
    }

    fn flush(&mut self) -> std::io::Result<()> {
        self.writer.flush()
    }
}

impl From<BodyHandle> for Body {
    fn from(handle: BodyHandle) -> Self {
        // we clone the handle here in order to have an owned type for the reader and writer, but
        // this means we have to be careful that we don't make the aliasing observable from the
        // public interface
        let handle2 = unsafe { handle.handle() };
        Self {
            reader: BufReader::new(handle),
            writer: BufWriter::new(handle2),
        }
    }
}

impl TryFrom<&str> for Body {
    type Error = Error;

    fn try_from(s: &str) -> Result<Self, Error> {
        Ok(BodyHandle::try_from(s)?.into())
    }
}

impl TryFrom<String> for Body {
    type Error = Error;

    fn try_from(s: String) -> Result<Self, Error> {
        Ok(BodyHandle::try_from(s)?.into())
    }
}

impl TryFrom<&[u8]> for Body {
    type Error = Error;

    fn try_from(s: &[u8]) -> Result<Self, Error> {
        Ok(BodyHandle::try_from(s)?.into())
    }
}

impl TryFrom<Vec<u8>> for Body {
    type Error = Error;

    fn try_from(s: Vec<u8>) -> Result<Self, Error> {
        Ok(BodyHandle::try_from(s)?.into())
    }
}