Compute transfers.

src/flow/flow.rs

@@ -1,5 +1,5 @@
 use crate::flow::adjacencies::Adjacencies;
-use crate::flow::Node;
+use crate::flow::{node_as_address, node_as_token_edge, Node};
 use crate::types::{Address, Edge, U256};
 use std::cmp::min;
 use std::collections::HashMap;
@@ -43,6 +43,8 @@ pub fn compute_flow(
         }
     }
     println!("Max flow: {flow}");
+    let transfers = extract_transfers(source, sink, &flow, used_edges);
+    println!("Num transfers: {}", transfers.len());
     flow.to_string()
 }
 
@@ -87,3 +89,88 @@ fn trace(parent: HashMap<Node, Node>, source: &Node, sink: &Node) -> Vec<Node> {
     }
     t
 }
+
+fn extract_transfers(
+    source: &Address,
+    sink: &Address,
+    amount: &U256,
+    mut used_edges: HashMap<Node, HashMap<Node, U256>>,
+) -> Vec<Edge> {
+    let mut transfers: Vec<Edge> = Vec::new();
+    let mut account_balances: HashMap<Address, U256> = HashMap::new();
+    account_balances.insert(*source, amount.clone());
+
+    while !account_balances.is_empty()
+        && (account_balances.len() > 1 || *account_balances.iter().nth(0).unwrap().0 != *sink)
+    {
+        let next = extract_next_transfers(&mut used_edges, &mut account_balances);
+        assert!(!next.is_empty());
+        transfers.extend(next.into_iter());
+    }
+
+    transfers
+}
+
+/// Extract the next list of transfers until we get to a situation where
+/// we cannot transfer the full balance and start over.
+fn extract_next_transfers(
+    used_edges: &mut HashMap<Node, HashMap<Node, U256>>,
+    account_balances: &mut HashMap<Address, U256>,
+) -> Vec<Edge> {
+    let mut transfers = Vec::new();
+
+    loop {
+        let first_edge = transfers.is_empty();
+        if let Some(edge) = next_nonzero_edge(used_edges, account_balances, first_edge) {
+            account_balances
+                .entry(edge.from)
+                .and_modify(|balance| *balance -= edge.capacity);
+            account_balances
+                .entry(edge.to)
+                .and_modify(|balance| *balance += edge.capacity);
+            account_balances.retain(|_account, balance| balance > &mut U256::from(0));
+            used_edges
+                .entry(Node::Node(edge.from))
+                .and_modify(|outgoing| {
+                    outgoing.remove(&Node::TokenEdge(edge.from, edge.token));
+                });
+            transfers.push(edge);
+        } else {
+            return transfers;
+        }
+    }
+}
+
+fn next_nonzero_edge(
+    used_edges: &HashMap<Node, HashMap<Node, U256>>,
+    account_balances: &HashMap<Address, U256>,
+    first_edge: bool,
+) -> Option<Edge> {
+    for (account, balance) in account_balances {
+        for (intermediate, _) in &used_edges[&Node::Node(*account)] {
+            let (from, token) = node_as_token_edge(intermediate);
+            for (to_node, capacity) in &used_edges[intermediate] {
+                let to = node_as_address(to_node);
+                if *capacity == U256::from(0) {
+                    continue;
+                }
+                if *balance < *capacity {
+                    // We do not have enough balance yet, there will be another transfer along this edge.
+                    if first_edge {
+                        continue;
+                    } else {
+                        return None;
+                    }
+                } else {
+                    return Some(Edge {
+                        from: *from,
+                        to: *to,
+                        token: *token,
+                        capacity: *capacity,
+                    });
+                }
+            }
+        }
+    }
+    None
+}

src/flow/mod.rs

@@ -10,6 +10,22 @@ pub enum Node {
     TokenEdge(Address, Address),
 }
 
+pub fn node_as_address(node: &Node) -> &Address {
+    if let Node::Node(address) = node {
+        address
+    } else {
+        panic!()
+    }
+}
+
+pub fn node_as_token_edge(node: &Node) -> (&Address, &Address) {
+    if let Node::TokenEdge(from, token) = node {
+        (from, token)
+    } else {
+        panic!()
+    }
+}
+
 impl Display for Node {
     fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
         match self {

src/server.rs

@@ -83,21 +83,6 @@ impl Server {
 }
 
 fn read_request(socket: &mut TcpStream) -> Result<JsonRpcRequest, Box<dyn Error>> {
-    // let mut buf_reader = BufReader::new(&mut socket);
-    // let http_request: Vec<_> = buf_reader
-    //     .by_ref()
-    //     .lines()
-    //     .map(|result| result.unwrap())
-    //     .take_while(|line| !line.is_empty())
-    //     .collect();
-    // println!("{http_request:?}");
-    // let mut buf = [0; 74];
-    // buf_reader.read_exact(&mut buf)?;
-    // println!("payload: {buf:?}");
-
-    // let response = "HTTP/1.1 200 OK\r\n\r\n";
-
-    // socket.write_all(response.as_bytes()).unwrap();
     let payload = read_payload(socket)?;
     let mut request = json::parse(&String::from_utf8(payload)?)?;
     println!("Request: {request}");