request.rs 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162
  1. use std::future::Future;
  2. use crate::{
  3. error::{InternalError, SystemError},
  4. module::Event,
  5. request::payload::Payload,
  6. response::Responder,
  7. util::ready::{ready, Ready},
  8. };
  9. use futures_core::ready;
  10. use std::{
  11. fmt::Debug,
  12. hash::Hash,
  13. ops,
  14. pin::Pin,
  15. task::{Context, Poll},
  16. };
  17. #[derive(Clone, Debug)]
  18. pub struct EventRequest {
  19. pub(crate) id: String,
  20. pub(crate) event: Event,
  21. }
  22. impl EventRequest {
  23. pub fn new<E>(event: E) -> EventRequest
  24. where
  25. E: Into<Event>,
  26. {
  27. Self {
  28. id: uuid::Uuid::new_v4().to_string(),
  29. event: event.into(),
  30. }
  31. }
  32. }
  33. pub trait FromRequest: Sized {
  34. type Error: Into<SystemError>;
  35. type Future: Future<Output = Result<Self, Self::Error>>;
  36. fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future;
  37. }
  38. #[doc(hidden)]
  39. impl FromRequest for () {
  40. type Error = SystemError;
  41. type Future = Ready<Result<(), SystemError>>;
  42. fn from_request(_req: &EventRequest, _payload: &mut Payload) -> Self::Future { ready(Ok(())) }
  43. }
  44. #[doc(hidden)]
  45. impl FromRequest for String {
  46. type Error = SystemError;
  47. type Future = Ready<Result<Self, Self::Error>>;
  48. fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future {
  49. match &payload {
  50. Payload::None => ready(Err(unexpected_none_payload(req))),
  51. Payload::Bytes(buf) => ready(Ok(String::from_utf8_lossy(buf).into_owned())),
  52. }
  53. }
  54. }
  55. fn unexpected_none_payload(request: &EventRequest) -> SystemError {
  56. log::warn!(
  57. "Event: {:?} expected payload but payload is empty",
  58. &request.event
  59. );
  60. InternalError::new("Expected payload but payload is empty").into()
  61. }
  62. #[doc(hidden)]
  63. impl<T> FromRequest for Result<T, T::Error>
  64. where
  65. T: FromRequest,
  66. {
  67. type Error = SystemError;
  68. type Future = FromRequestFuture<T::Future>;
  69. fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future {
  70. FromRequestFuture {
  71. fut: T::from_request(req, payload),
  72. }
  73. }
  74. }
  75. #[pin_project::pin_project]
  76. pub struct FromRequestFuture<Fut> {
  77. #[pin]
  78. fut: Fut,
  79. }
  80. impl<Fut, T, E> Future for FromRequestFuture<Fut>
  81. where
  82. Fut: Future<Output = Result<T, E>>,
  83. {
  84. type Output = Result<Result<T, E>, SystemError>;
  85. fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
  86. let this = self.project();
  87. let res = ready!(this.fut.poll(cx));
  88. Poll::Ready(Ok(res))
  89. }
  90. }
  91. pub struct In<T>(pub T);
  92. impl<T> In<T> {
  93. pub fn into_inner(self) -> T { self.0 }
  94. }
  95. impl<T> ops::Deref for In<T> {
  96. type Target = T;
  97. fn deref(&self) -> &T { &self.0 }
  98. }
  99. impl<T> ops::DerefMut for In<T> {
  100. fn deref_mut(&mut self) -> &mut T { &mut self.0 }
  101. }
  102. #[cfg(feature = "use_serde")]
  103. impl<T> FromRequest for In<T>
  104. where
  105. T: serde::de::DeserializeOwned + 'static,
  106. {
  107. type Error = SystemError;
  108. type Future = Ready<Result<Self, SystemError>>;
  109. #[inline]
  110. fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future {
  111. match payload {
  112. Payload::None => ready(Err(unexpected_none_payload(req))),
  113. Payload::Bytes(bytes) => {
  114. let data: T = bincode::deserialize(bytes).unwrap();
  115. ready(Ok(In(data)))
  116. },
  117. }
  118. }
  119. }
  120. #[cfg(feature = "use_protobuf")]
  121. impl<T> FromRequest for In<T>
  122. where
  123. T: ::protobuf::Message + 'static,
  124. {
  125. type Error = SystemError;
  126. type Future = Ready<Result<Self, SystemError>>;
  127. #[inline]
  128. fn from_request(req: &EventRequest, payload: &mut Payload) -> Self::Future {
  129. match payload {
  130. Payload::None => ready(Err(unexpected_none_payload(req))),
  131. Payload::Bytes(bytes) => {
  132. let data: T = ::protobuf::Message::parse_from_bytes(bytes).unwrap();
  133. ready(Ok(In(data)))
  134. },
  135. }
  136. }
  137. }