Implement WhonixTorControlReader

network/tor/tor/src/main/java/bisq/tor/controller/WhonixTorControlReader.java

@@ -0,0 +1,74 @@
+package bisq.tor.controller;
+
+import bisq.tor.controller.events.events.BootstrapEvent;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+@Slf4j
+public class WhonixTorControlReader implements AutoCloseable {
+    private final BufferedReader bufferedReader;
+    private final BlockingQueue<String> replies = new LinkedBlockingQueue<>();
+    private Optional<Thread> workerThread = Optional.empty();
+
+    public WhonixTorControlReader(InputStream inputStream) {
+        bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.US_ASCII));
+    }
+
+    public void start() {
+        Thread thread = new Thread(() -> {
+            try {
+                String line;
+                while ((line = bufferedReader.readLine()) != null) {
+
+                    if (isStatusClientEvent(line)) {
+                        Optional<BootstrapEvent> bootstrapEvent = BootstrapEventParser.tryParse(line);
+
+                        if (bootstrapEvent.isPresent()) {
+                            log.info("{}", bootstrapEvent.get());
+                        } else {
+                            log.info("Unknown status client event: {}", line);
+                        }
+
+                    } else {
+                        replies.add(line);
+                    }
+
+                    if (Thread.interrupted()) {
+                        break;
+                    }
+                }
+            } catch (IOException e) {
+                log.error("Tor control port reader couldn't read reply.", e);
+            }
+
+        });
+        workerThread = Optional.of(thread);
+        thread.start();
+    }
+
+    @Override
+    public void close() throws Exception {
+        workerThread.ifPresent(Thread::interrupt);
+    }
+
+    public String readLine() {
+        try {
+            return replies.take();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private boolean isStatusClientEvent(String line) {
+        // 650 STATUS_CLIENT NOTICE CIRCUIT_ESTABLISHED
+        return line.startsWith("650 STATUS_CLIENT");
+    }
+}

network/tor/tor/src/main/java/bisq/tor/controller/WhonixTorController.java

@@ -11,15 +11,12 @@
 
 public class WhonixTorController implements AutoCloseable {
     private final Socket controlSocket;
-    private final BufferedReader bufferedReader;
+    private final WhonixTorControlReader whonixTorControlReader;
     private final OutputStream outputStream;
 
     public WhonixTorController() throws IOException {
         controlSocket = new Socket("127.0.0.1", 9051);
-
-        InputStream inputStream = controlSocket.getInputStream();
-        bufferedReader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.US_ASCII));
-
+        whonixTorControlReader = new WhonixTorControlReader(controlSocket.getInputStream());
         outputStream = controlSocket.getOutputStream();
     }
 
@@ -28,6 +25,10 @@ public void close() throws IOException {
         controlSocket.close();
     }
 
+    public void initialize() {
+        whonixTorControlReader.start();
+    }
+
     public void authenticate(PasswordDigest passwordDigest) throws IOException {
         byte[] secret = passwordDigest.getSecret();
         String secretHex = Hex.encode(secret);
@@ -99,8 +100,8 @@ private void sendCommand(String command) throws IOException {
         outputStream.flush();
     }
 
-    private String receiveReply() throws IOException {
-        String reply = bufferedReader.readLine();
+    private String receiveReply() {
+        String reply = whonixTorControlReader.readLine();
         if (reply.equals("510 Command filtered")) {
             throw new TorCommandFilteredException();
         }