| 108 | } |
| 109 | |
| 110 | async fn handle_stream( |
| 111 | stream_id: &str, |
| 112 | channel: &mpsc::Sender<api::LogItem>, |
| 113 | k: &str, |
| 114 | v: &redis::Value, |
| 115 | ) -> Result<()> { |
| 116 | match k { |
| 117 | "up" => { |
| 118 | trace!(key = %k, id = %stream_id, "Event-log received from stream"); |
| 119 | if let redis::Value::BulkString(b) = v { |
| 120 | let pl = integration::UplinkEvent::decode(&mut Cursor::new(b))?; |
| 121 | let pl = api::LogItem { |
| 122 | id: stream_id.to_string(), |
| 123 | time: pl.time.as_ref().map(|v| prost_types::Timestamp { |
| 124 | seconds: v.seconds, |
| 125 | nanos: v.nanos, |
| 126 | }), |
| 127 | description: k.to_string(), |
| 128 | body: serde_json::to_string(&pl)?, |
| 129 | properties: [ |
| 130 | ("DR".to_string(), pl.dr.to_string()), |
| 131 | ("FPort".to_string(), pl.f_port.to_string()), |
| 132 | ("FCnt".to_string(), pl.f_cnt.to_string()), |
| 133 | ("Data".to_string(), hex::encode(&pl.data)), |
| 134 | ] |
| 135 | .iter() |
| 136 | .cloned() |
| 137 | .collect(), |
| 138 | }; |
| 139 | |
| 140 | channel.send(pl).await?; |
| 141 | } |
| 142 | } |
| 143 | "join" => { |
| 144 | trace!(key = %k, id = %stream_id, "Event-log received from stream"); |
| 145 | if let redis::Value::BulkString(b) = v { |
| 146 | let pl = integration::JoinEvent::decode(&mut Cursor::new(b))?; |
| 147 | let pl = api::LogItem { |
| 148 | id: stream_id.to_string(), |
| 149 | time: pl.time.as_ref().map(|v| prost_types::Timestamp { |
| 150 | seconds: v.seconds, |
| 151 | nanos: v.nanos, |
| 152 | }), |
| 153 | description: k.to_string(), |
| 154 | body: serde_json::to_string(&pl)?, |
| 155 | properties: [("DevAddr".to_string(), pl.dev_addr)] |
| 156 | .iter() |
| 157 | .cloned() |
| 158 | .collect(), |
| 159 | }; |
| 160 | |
| 161 | channel.send(pl).await?; |
| 162 | } |
| 163 | } |
| 164 | "ack" => { |
| 165 | trace!(key = %k, id = %stream_id, "Event-log received from stream"); |
| 166 | if let redis::Value::BulkString(b) = v { |
| 167 | let pl = integration::AckEvent::decode(&mut Cursor::new(b))?; |