GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #7 from nymtech/feature/provider_type

Browse Source 
Provider storing received messages
This commit is contained in:
Dave Hrycyszyn
2019-12-10 16:00:12 +00:00
committed by GitHub
parent 86861ccb70 528ba87492
commit 8d0afbd21e
4 changed files with 335 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+123 -6
View File
@@ -240,11 +240,88 @@ name = "fuchsia-zircon-sys"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-channel"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-core"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-executor"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-io"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-macro"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "futures-sink"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-task"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "futures-util"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"memchr 2.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
"pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)",
"proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
"slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "generic-array"
version = "0.12.3"
@@ -271,6 +348,11 @@ dependencies = [
"libc 0.2.65 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "hex"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "hkdf"
version = "0.8.0"
@@ -429,8 +511,11 @@ version = "0.1.0"
dependencies = [
"clap 2.33.0 (registry+https://github.com/rust-lang/crates.io-index)",
"curve25519-dalek 1.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
"futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)",
"rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
"sphinx 0.1.0",
"tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
"tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
@@ -443,11 +528,31 @@ name = "pin-project-lite"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "pin-utils"
version = "0.1.0-alpha.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "ppv-lite86"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "proc-macro-hack"
version = "0.5.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
name = "proc-macro-nested"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "proc-macro2"
version = "1.0.6"
@@ -628,7 +733,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
[[package]]
name = "syn"
version = "1.0.9"
version = "1.0.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -646,7 +751,7 @@ dependencies = [
[[package]]
name = "tokio"
version = "0.2.1"
version = "0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"bytes 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -672,7 +777,7 @@ version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
dependencies = [
"quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)",
"syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)",
]
[[package]]
@@ -770,10 +875,19 @@ dependencies = [
"checksum fuchsia-cprng 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba"
"checksum fuchsia-zircon 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82"
"checksum fuchsia-zircon-sys 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7"
"checksum futures 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b6f16056ecbb57525ff698bb955162d0cd03bee84e6241c27ff75c08d8ca5987"
"checksum futures-channel 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "fcae98ca17d102fd8a3603727b9259fcf7fa4239b603d2142926189bc8999b86"
"checksum futures-core 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "79564c427afefab1dfb3298535b21eda083ef7935b4f0ecbfcb121f0aec10866"
"checksum futures-executor 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "1e274736563f686a837a0568b478bdabfeaec2dca794b5649b04e2fe1627c231"
"checksum futures-io 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "e676577d229e70952ab25f3945795ba5b16d63ca794ca9d2c860e5595d20b5ff"
"checksum futures-macro 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "52e7c56c15537adb4f76d0b7a76ad131cb4d2f4f32d3b0bcabcbe1c7c5e87764"
"checksum futures-sink 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "171be33efae63c2d59e6dbba34186fe0d6394fb378069a76dfd80fdcffd43c16"
"checksum futures-task 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "0bae52d6b29cf440e298856fec3965ee6fa71b06aa7495178615953fd669e5f9"
"checksum futures-util 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "c0d66274fb76985d3c62c886d1da7ac4c0903a8c9f754e8fe0f35a6a6cc39e76"
"checksum generic-array 0.12.3 (registry+https://github.com/rust-lang/crates.io-index)" = "c68f0274ae0e023facc3c97b2e00f076be70e254bc851d972503b328db79b2ec"
"checksum getrandom 0.1.13 (registry+https://github.com/rust-lang/crates.io-index)" = "e7db7ca94ed4cd01190ceee0d8a8052f08a247aa1b469a7f68c6a3b71afcf407"
"checksum hermit-abi 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "307c3c9f937f38e3534b1d6447ecf090cafcc9744e4a6360e8b037b2cf5af120"
"checksum hex 0.4.0 (registry+https://github.com/rust-lang/crates.io-index)" = "023b39be39e3a2da62a94feb433e91e8bcd37676fbc8bea371daf52b7a769a3e"
"checksum hkdf 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "3fa08a006102488bd9cd5b8013aabe84955cf5ae22e304c2caf655b633aefae3"
"checksum hmac 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "5dcb5e64cda4c23119ab41ba960d1e170a774c8e4b9d9e6a9bc18aabf5e59695"
"checksum iovec 0.1.4 (registry+https://github.com/rust-lang/crates.io-index)" = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e"
@@ -793,7 +907,10 @@ dependencies = [
"checksum num_cpus 1.11.1 (registry+https://github.com/rust-lang/crates.io-index)" = "76dac5ed2a876980778b8b85f75a71b6cbf0db0b1232ee12f826bccb00d09d72"
"checksum opaque-debug 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "2839e79665f131bdb5782e51f2c6c9599c133c6098982a54c794358bf432529c"
"checksum pin-project-lite 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f0af6cbca0e6e3ce8692ee19fb8d734b641899e07b68eb73e9bbbd32f1703991"
"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587"
"checksum ppv-lite86 0.2.6 (registry+https://github.com/rust-lang/crates.io-index)" = "74490b50b9fbe561ac330df47c08f3f33073d2d00c150f719147d7c54522fa1b"
"checksum proc-macro-hack 0.5.11 (registry+https://github.com/rust-lang/crates.io-index)" = "ecd45702f76d6d3c75a80564378ae228a85f0b59d2f3ed43c91b4a69eb2ebfc5"
"checksum proc-macro-nested 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "369a6ed065f249a159e06c45752c780bda2fb53c995718f9e484d08daa9eb42e"
"checksum proc-macro2 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "9c9e470a8dc4aeae2dee2f335e8f533e2d4b347e1434e5671afc49b054592f27"
"checksum quote 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "053a8c8bcc71fcce321828dc897a98ab9760bef03a4fc36693c231e5b3216cfe"
"checksum rand 0.7.2 (registry+https://github.com/rust-lang/crates.io-index)" = "3ae1b169243eaf61759b8475a998f0a385e42042370f3a7dbaf35246eacc8412"
@@ -814,9 +931,9 @@ dependencies = [
"checksum strsim 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "8ea5119cdb4c55b55d432abb513a0429384878c15dde60cc77b1c99de1a95a6a"
"checksum subtle 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2d67a5a62ba6e01cb2192ff309324cb4875d0c451d55fe2319433abe7a05a8ee"
"checksum subtle 2.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7c65d530b10ccaeac294f349038a597e435b18fb456aadd0840a623f83b9e941"
"checksum syn 1.0.9 (registry+https://github.com/rust-lang/crates.io-index)" = "f89693ae015201f8de93fd96bde2d065f8bfc3f97ce006d5bc9f900b97c0c7c0"
"checksum syn 1.0.11 (registry+https://github.com/rust-lang/crates.io-index)" = "dff0acdb207ae2fe6d5976617f887eb1e35a2ba52c13c7234c790960cdad9238"
"checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060"
"checksum tokio 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "beeef686ef92a222de07e089f455d9f8478bbba9651718f9e4b276babe829082"
"checksum tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)" = "bcced6bb623d4bff3739c176c415f13c418f426395c169c9c3cd9a492c715b16"
"checksum tokio-macros 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d5795a71419535c6dcecc9b6ca95bdd3c2d6142f7e8343d7beb9923f129aa87e"
"checksum typenum 1.11.2 (registry+https://github.com/rust-lang/crates.io-index)" = "6d2783fe2d6b8c1101136184eb41be8b1ad379e4657050b8aaff0c79ee7575f9"
"checksum unicode-width 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "caaa9d531767d1ff2150b9332433f32a24622147e5ebb1f26409d5da67afd479"
Cargo.toml
+6 -3
View File
@@ -7,7 +7,10 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
sphinx = { path = "../sphinx" }
tokio = { version = "0.2", features = ["full"] }
clap = "2.33.0"
curve25519-dalek = "1.2.3"
clap = "2.33.0"
hex = "0.4.0"
futures = "0.3.1"
rand = "0.7.2"
sphinx = { path = "../sphinx" }
tokio = { version = "0.2.4", features = ["full"] }
src/main.rs
+16 -2
View File
@@ -1,6 +1,8 @@
use crate::provider::ServiceProvider;
use clap::{App, Arg, ArgMatches, SubCommand};
use curve25519_dalek::scalar::Scalar;
use std::net::ToSocketAddrs;
use std::path::PathBuf;
use std::process;
mod provider;
@@ -33,9 +35,12 @@ fn run(matches: &ArgMatches) {
}
};
let store_dir = PathBuf::from(matches.value_of("storeDir").unwrap());
println!("The value of host is: {:?}", host);
println!("The value of port is: {:?}", port);
println!("The value of key is: {:?}", secret_key);
println!("The value of storeDir is: {:?}", store_dir);
let socket_address = (host, port)
.to_socket_addrs()
@@ -48,7 +53,8 @@ fn run(matches: &ArgMatches) {
// make sure our socket_address is equal to our predefined-hardcoded value
assert_eq!("127.0.0.1:8081", socket_address.to_string());
provider::listening_loop().unwrap();
let provider = ServiceProvider::new(socket_address, secret_key, store_dir);
provider.start_listening().unwrap()
}
fn main() {
@@ -58,7 +64,7 @@ fn main() {
.about("Implementation of the Loopix-based Service Provider")
.subcommand(
SubCommand::with_name("run")
.about("Starts the mixnode")
.about("Starts the service provider")
.arg(
Arg::with_name("host")
.short("h")
@@ -80,6 +86,14 @@ fn main() {
.long("keyfile")
.help("Optional path to the persistent keyfile of the node")
.takes_value(true),
)
.arg(
Arg::with_name("storeDir")
.short("sdir")
.long("storeDir")
.help("Directory storing all packets for the clients")
.takes_value(true)
.required(true),
),
)
.get_matches();
src/provider/mod.rs
+190 -32
View File
@@ -1,47 +1,205 @@
use std::fs::File;
use std::io::Write;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::RwLock;
use curve25519_dalek::scalar::Scalar;
use rand::Rng;
use sphinx::{ProcessedPacket, SphinxPacket};
use sphinx::route::{DestinationAddressBytes, SURBIdentifier};
use tokio::prelude::*;
use tokio::runtime::Runtime;
pub fn listening_loop() -> Result<(), Box<dyn std::error::Error>> {
// Create the runtime, probably later move it to Provider struct itself?
let mut rt = Runtime::new()?;
// TODO: if we ever create config file, this should go there
const STORED_MESSAGE_FILENAME_LENGTH: usize = 16;
// Spawn the root task
rt.block_on(async {
let my_address = "127.0.0.1:8081";
let mut listener = tokio::net::TcpListener::bind(my_address).await?;
println!("Starting Nym store-and-forward Provider on address {:?}", my_address);
println!("Waiting for input...");
// TODO: this will probably need to be moved elsewhere I imagine
// DUPLICATE WITH MIXNODE CODE!!!
#[derive(Debug)]
pub enum MixProcessingError {
SphinxRecoveryError,
ReceivedForwardHopError,
InvalidPayload,
NonMatchingRecipient,
FileIOFailure,
}
impl From<sphinx::ProcessingError> for MixProcessingError {
// for time being just have a single error instance for all possible results of sphinx::ProcessingError
fn from(_: sphinx::ProcessingError) -> Self {
use MixProcessingError::*;
SphinxRecoveryError
}
}
impl From<std::io::Error> for MixProcessingError {
fn from(_: std::io::Error) -> Self {
use MixProcessingError::*;
FileIOFailure
}
}
// ProcessingData defines all data required to correctly unwrap sphinx packets
// Do note that we're copying this struct around and hence the secret_key.
// It might, or might not be, what we want
#[derive(Debug, Clone)]
struct ProcessingData {
secret_key: Scalar,
store_dir: PathBuf,
}
impl ProcessingData {
fn new(secret_key: Scalar, store_dir: PathBuf) -> Self {
ProcessingData {
secret_key,
store_dir,
}
}
fn add_arc_rwlock(self) -> Arc<RwLock<Self>> {
Arc::new(RwLock::new(self))
}
}
struct StoreData {
client_address: DestinationAddressBytes,
client_surb_id: SURBIdentifier,
message: Vec<u8>,
}
struct PacketProcessor(());
impl PacketProcessor {
fn process_sphinx_data_packet(packet_data: &[u8], processing_data: &RwLock<ProcessingData>) -> Result<StoreData, MixProcessingError> {
let packet = SphinxPacket::from_bytes(packet_data.to_vec())?;
let read_processing_data = processing_data.read().unwrap();
let (client_address, client_surb_id, payload) = match packet.process(read_processing_data.secret_key) {
ProcessedPacket::ProcessedPacketFinalHop(client_address, surb_id, payload) => (client_address, surb_id, payload),
_ => return Err(MixProcessingError::ReceivedForwardHopError),
};
let (payload_destination, message) = payload.try_recover_destination_and_plaintext().ok_or_else(|| MixProcessingError::InvalidPayload)?;
if client_address != payload_destination {
return Err(MixProcessingError::NonMatchingRecipient);
}
Ok(StoreData {
client_address,
client_surb_id,
message,
})
}
fn generate_random_file_name() -> String {
rand::thread_rng().sample_iter(&rand::distributions::Alphanumeric).take(STORED_MESSAGE_FILENAME_LENGTH).collect::<String>()
}
fn store_processed_data(store_data: StoreData, store_dir: &Path) -> Result<(), MixProcessingError> {
let client_dir_name = hex::encode(store_data.client_address);
let full_store_dir = store_dir.join(client_dir_name);
let full_store_path = full_store_dir.join(PacketProcessor::generate_random_file_name());
println!("going to store: {:?} in file: {:?}", store_data.message, full_store_path);
// TODO: what to do with surbIDs??
// we can use normal io here, no need for tokio as it's all happening in one thread per connection
std::fs::create_dir_all(full_store_dir)?;
let mut file = File::create(full_store_path)?;
file.write_all(store_data.message.as_ref())?;
Ok(())
}
}
pub struct ServiceProvider {
network_address: SocketAddr,
secret_key: Scalar,
store_dir: PathBuf,
}
impl ServiceProvider {
pub fn new(network_address: SocketAddr, secret_key: Scalar, store_dir: PathBuf) -> Self {
ServiceProvider {
network_address,
secret_key,
store_dir,
}
}
async fn process_socket_connection(mut socket: tokio::net::TcpStream, processing_data: Arc<RwLock<ProcessingData>>) {
// TODO: we will actually need to distinguish multiple types of requests here;
// either from mixnodes to store final hop information
// or from clients to pull messages
let mut buf = [0u8; sphinx::PACKET_SIZE];
// In a loop, read data from the socket and write the data back.
loop {
let (mut inbound, _) = listener.accept().await?;
tokio::spawn(async move {
let mut buf = [0; sphinx::PACKET_SIZE];
loop {
match inbound.read(&mut buf).await {
Ok(length) if length == 0 =>
{
println!("Remote connection closed.");
return;
}
Ok(_) => {
let packet = SphinxPacket::from_bytes(buf.to_vec()).unwrap();
let payload = match packet.process(Default::default()) {
ProcessedPacket::ProcessedPacketFinalHop(_, _, payload) => Some(payload),
_ => None,
}.unwrap();
let message = payload.get_content();
}
match socket.read(&mut buf).await {
// socket closed
Ok(n) if n == 0 => {
println!("Remote connection closed.");
return;
}
Ok(_) => {
let store_data = match PacketProcessor::process_sphinx_data_packet(buf.as_ref(), processing_data.as_ref()) {
Ok(sd) => sd,
Err(e) => {
println!("failed to read from socket; err = {:?}", e);
eprintln!("failed to process sphinx packet; err = {:?}", e);
return;
}
};
PacketProcessor::store_processed_data(store_data, processing_data.read().unwrap().store_dir.as_path()).unwrap_or_else(|e| {
eprintln!("failed to store processed sphinx message; err = {:?}", e)
});
}
});
Err(e) => {
eprintln!("failed to read from socket; err = {:?}", e);
return;
}
};
// Write the some data back
if let Err(e) = socket.write_all(b"foomp").await {
eprintln!("failed to write reply to socket; err = {:?}", e);
return;
}
}
})
}
pub fn start_listening(&self) -> Result<(), Box<dyn std::error::Error>> {
// Create the runtime, probably later move it to Provider struct itself?
// TODO: figure out the difference between Runtime and Handle
let mut rt = Runtime::new()?;
// let mut h = rt.handle();
// Spawn the root task
rt.block_on(async {
let mut listener = tokio::net::TcpListener::bind(self.network_address).await?;
let processing_data = ProcessingData::new(self.secret_key, self.store_dir.clone()).add_arc_rwlock();
loop {
let (socket, _) = listener.accept().await?;
let thread_processing_data = processing_data.clone();
tokio::spawn(async move {
ServiceProvider::process_socket_connection(socket, thread_processing_data).await
});
}
})
}
}