backtalk 0.1.0

An asynchronous, streaming web server for JSON APIs
Documentation 
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

use {JsonValue, Request, Method, JsonObject, Error, channel};
use std::fmt;
use hyper::server as http;
use hyper::Error as HyperError;
use hyper::header::{ContentLength, ContentType};
use hyper::mime;
use hyper::Chunk as HyperChunk;
use futures::{Poll, Stream, Async, IntoFuture};
use futures::future::{ok, FutureResult, BoxFuture, Future};
use futures::stream::BoxStream;
use futures::sync::mpsc;
use Sender;

type ChunkReceiver = BoxStream<HyperChunk, ()>;

/**
A successful response with JSON data to be sent back to the client.

There are two kinds of replies. Static replies represent JSON data that is ready. Most requests
return static replies. Streaming replies represent a stream of JSON data that will stream from
a `Channel` directly to the client. You can't access the data of a streaming reply through the
`Reply` struct, since it's not ready yet. If you want to transform or edit the reply data for a
stream, you'll need to implement a custom `Channel` instead.

These are several ways to create a Reply:

- pass a Request to an Adapter to get a static response from a database
- pass a Request to a Channel to get a streaming response
- in your custom Resource, call `request.into_reply(data)` to create a Reply object.

Reply implements `IntoFuture`, so you can return it directly from a `and_then` block.
*/
#[derive(Debug)]
pub struct Reply {
  data: ReplyData,
  req: Request,
}

enum ReplyData {
  Value(JsonObject),
  Stream(ChunkReceiver),
}

impl fmt::Debug for ReplyData {
  fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
    match self {
      &ReplyData::Value(ref val) => write!(f, "ReplyData::Value({:?})", val),
      &ReplyData::Stream(_) => write!(f, "ReplyData::Stream(<stream>)"),
    }
  }
}

// only used internally
pub fn make_reply(req: Request, data: JsonObject) -> Reply {
  Reply {
    req: req,
    data: ReplyData::Value(data),
  }
}

// only used internally
pub fn make_streamed_reply(req: Request) -> (Sender, Reply) {
  let (tx, rx) = mpsc::unbounded();
  let rx = rx
    .map(|val: (String, JsonObject)| -> HyperChunk {
      format!("event:{}\ndata:{}\n\n", val.0, JsonValue::Object(val.1)).into()
    })
    .boxed();
  let reply = Reply {
    req: req,
    data: ReplyData::Stream(rx)
  };
  let sender = channel::new_sender(tx);
  (sender, reply)
}

impl Reply {
  pub fn data(&self) -> Option<&JsonObject> {
    match self.data {
      ReplyData::Value(ref dat) => Some(dat),
      _ => None,
    }
  }

  pub fn data_mut(&mut self) -> Option<&mut JsonObject> {
    match self.data {
      ReplyData::Value(ref mut dat) => Some(dat),
      _ => None,
    }
  }

  // TODO data_then accepts a function that returns a future<JsonObject, Error>

  pub fn to_http(self) -> http::Response<Body> {
    let resp = http::Response::new();

    match self.data {
      ReplyData::Value(val) => {
        let resp_str = JsonValue::Object(val).to_string();
        resp
          .with_header(ContentLength(resp_str.len() as u64))
          .with_header(ContentType(mime::APPLICATION_JSON))
          .with_body(Body::Once(Some(resp_str.into())))
      },
      ReplyData::Stream(stream) => {
        resp
          .with_header(ContentType(mime::TEXT_EVENT_STREAM))
          .with_body(Body::Stream(stream))
      },
    }
  }

  pub fn method(&self) -> Method {
    self.req.method()
  }

  pub fn resource(&self) -> &str {
    &self.req.resource()
  }

  pub fn id(&self) -> &Option<String> {
    &self.req.id()
  }

  pub fn params(&self) -> &JsonObject {
    &self.req.params()
  }

  pub fn param(&self, key: &str) -> &JsonValue {
    self.req.param(key)
  }

  pub fn boxed(self) -> BoxFuture<Reply, Error> {
    ok(self).boxed()
  }

  pub fn request_data(&self) -> &JsonObject {
    self.req.data()
  }
}

impl IntoFuture for Reply {
  type Item = Reply;
  type Error = Error;
  type Future = FutureResult<Reply, Error>;
  fn into_future(self) -> Self::Future {
    ok(self)
  }
}

/// A `Stream` for `HyperChunk`s used in requests and responses.
pub enum Body {
  Once(Option<HyperChunk>),
  Stream(ChunkReceiver),
}

impl Stream for Body {
  type Item = HyperChunk;
  type Error = HyperError;

  fn poll(&mut self) -> Poll<Option<HyperChunk>, HyperError> {
    match self {
      &mut Body::Once(ref mut opt) => Ok(Async::Ready(opt.take())),
      &mut Body::Stream(ref mut stream) => {
        match stream.poll() {
          Ok(u) => Ok(u),
          Err(()) => Ok(Async::Ready(None)), // this probably can never happen?
        }
      }
    }
  }
}