wick-rpc 0.2.0

Library for implementing or interacting with a wick runtime via gRPC
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

/// Conversion utilities for RPC data structures.
pub(crate) mod conversions;
use std::time::Duration;

use serde::{Deserialize, Serialize};
pub use wick_interface_types::*;
use wick_packet::{Metadata, Packet, PacketError, PacketPayload, WickMetadata};

use crate::rpc::{self, packet as rpc_packet, Packet as RpcPacket};

/// Important statistics for the hosted components.
#[derive(Debug, Default, Clone, Serialize, Deserialize, PartialEq)]
#[non_exhaustive]
pub struct Statistics {
  /// The name of the component.
  pub name: String,
  /// The number of times a component has been called.
  pub runs: u32,
  /// The number of times the component resulted in an unrecoverable error.
  pub errors: u32,
  /// Execution duration statistics.
  pub execution_duration: Option<DurationStatistics>,
}

mod as_micros {
  use std::convert::TryInto;
  use std::time::Duration;

  use serde::{Deserialize, Deserializer, Serializer};

  pub(crate) fn serialize<S>(duration: &Duration, serializer: S) -> Result<S::Ok, S::Error>
  where
    S: Serializer,
  {
    serializer.serialize_u64(duration.as_micros().try_into().unwrap_or(u64::MAX))
  }
  pub(crate) fn deserialize<'de, D>(deserializer: D) -> Result<Duration, D::Error>
  where
    D: Deserializer<'de>,
  {
    let micros = u64::deserialize(deserializer)?;
    Ok(Duration::from_micros(micros))
  }
}

/// Duration related statistics.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
#[non_exhaustive]
pub struct DurationStatistics {
  /// The maximum duration.
  #[serde(with = "as_micros")]
  pub max_time: Duration,
  /// The minimum duration.
  #[serde(with = "as_micros")]
  pub min_time: Duration,
  /// The average duration.
  #[serde(with = "as_micros")]
  pub average_time: Duration,
  /// The total duration.
  #[serde(with = "as_micros")]
  pub total_time: Duration,
}

impl DurationStatistics {
  /// Creates a new [DurationStatistics] instance.
  #[must_use]
  pub const fn new(min_time: Duration, max_time: Duration, average_time: Duration, total_time: Duration) -> Self {
    Self {
      max_time,
      min_time,
      average_time,
      total_time,
    }
  }
}

impl RpcPacket {
  /// Converts a [RpcPacket] into a [Packet].
  pub fn into_packet(self) -> Packet {
    self.into()
  }
}

impl From<RpcPacket> for Packet {
  fn from(v: RpcPacket) -> Self {
    let (op, port, done) = v.metadata.map_or_else(
      || (0, Packet::FATAL_ERROR.to_owned(), 0_u8),
      |m| (m.index, m.port, m.flags.try_into().unwrap()),
    );
    Self::new_raw(
      v.data
        .map_or(PacketPayload::fatal_error("Could not decode RPC message"), |p| p.into()),
      Metadata::new(op),
      WickMetadata::new(port, done),
    )
  }
}

impl From<rpc_packet::Data> for PacketPayload {
  fn from(v: rpc_packet::Data) -> Self {
    match v {
      rpc_packet::Data::Ok(v) => PacketPayload::Ok(match v.data {
        Some(rpc::ok::Data::Messagepack(v)) => Some(v.into()),
        Some(rpc::ok::Data::Json(_v)) => todo!(),
        None => unreachable!(),
      }),

      rpc_packet::Data::Err(v) => PacketPayload::Err(PacketError::new(v.message)),
    }
  }
}

impl From<PacketPayload> for rpc_packet::Data {
  fn from(v: PacketPayload) -> Self {
    match v {
      PacketPayload::Ok(v) => rpc_packet::Data::Ok(rpc::Ok {
        data: Some(rpc::ok::Data::Messagepack(v.map_or_else(Vec::new, |v| v.to_vec()))),
      }),
      PacketPayload::Err(e) => rpc_packet::Data::Err(rpc::Err {
        message: e.msg().to_owned(),
        code: 513,
      }),
    }
  }
}