v1 datagram and raw support for sink

apps/sam/java/src/net/i2p/sam/client/SAMClientEventListenerImpl.java

@@ -14,6 +14,8 @@ public class SAMClientEventListenerImpl implements SAMReader.SAMClientEventListe
     public void streamConnectedReceived(String remoteDestination, String id) {}
     public void streamDataReceived(String id, byte[] data, int offset, int length) {}
     public void streamStatusReceived(String result, String id, String message) {}
+    public void datagramReceived(String dest, byte[] data, int offset, int length, int fromPort, int toPort) {}
+    public void rawReceived(byte[] data, int offset, int length, int fromPort, int toPort, int protocol) {}
     public void pingReceived(String data) {}
     public void pongReceived(String data) {}
     public void unknownMessageReceived(String major, String minor, Properties params) {}

apps/sam/java/src/net/i2p/sam/client/SAMReader.java

@@ -80,6 +80,8 @@ public class SAMReader {
         public void streamDataReceived(String id, byte data[], int offset, int length);
         public void namingReplyReceived(String name, String result, String value, String message);
         public void destReplyReceived(String publicKey, String privateKey);
+        public void datagramReceived(String dest, byte[] data, int offset, int length, int fromPort, int toPort);
+        public void rawReceived(byte[] data, int offset, int length, int fromPort, int toPort, int protocol);
         public void pingReceived(String data);
         public void pongReceived(String data);
         
@@ -231,6 +233,73 @@ public class SAMReader {
             } else {
                 _listener.unknownMessageReceived(major, minor, params);
             }
+        } else if ("DATAGRAM".equals(major)) {
+            if ("RECEIVED".equals(minor)) {
+                String dest = params.getProperty("DESTINATION");
+                String size = params.getProperty("SIZE");
+                String fp = params.getProperty("FROM_PORT");
+                String tp = params.getProperty("TO_PORT");
+                int fromPort = 0;
+                int toPort = 0;
+                if (dest != null) {
+                    try {
+                      if (fp != null)
+                          fromPort = Integer.parseInt(fp);
+                      if (tp != null)
+                          toPort = Integer.parseInt(tp);
+                        int sizeVal = Integer.parseInt(size);
+                        byte data[] = new byte[sizeVal];
+                        int read = DataHelper.read(_inRaw, data);
+                        if (read != sizeVal) {
+                            _listener.unknownMessageReceived(major, minor, params);
+                        } else {
+                            _listener.datagramReceived(dest, data, 0, sizeVal, fromPort, toPort);
+                        }
+                    } catch (NumberFormatException nfe) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    } catch (IOException ioe) {
+                        _live = false;
+                        _listener.unknownMessageReceived(major, minor, params);
+                    }
+                } else {
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
+        } else if ("RAW".equals(major)) {
+            if ("RECEIVED".equals(minor)) {
+                String size = params.getProperty("SIZE");
+                String fp = params.getProperty("FROM_PORT");
+                String tp = params.getProperty("TO_PORT");
+                String pr = params.getProperty("PROTOCOL");
+                int fromPort = 0;
+                int toPort = 0;
+                int protocol = 18;
+                try {
+                    if (fp != null)
+                        fromPort = Integer.parseInt(fp);
+                    if (tp != null)
+                        toPort = Integer.parseInt(tp);
+                    if (pr != null)
+                        protocol = Integer.parseInt(pr);
+                    int sizeVal = Integer.parseInt(size);
+                    byte data[] = new byte[sizeVal];
+                    int read = DataHelper.read(_inRaw, data);
+                    if (read != sizeVal) {
+                        _listener.unknownMessageReceived(major, minor, params);
+                    } else {
+                        _listener.rawReceived(data, 0, sizeVal, fromPort, toPort, protocol);
+                    }
+                } catch (NumberFormatException nfe) {
+                    _listener.unknownMessageReceived(major, minor, params);
+                } catch (IOException ioe) {
+                    _live = false;
+                    _listener.unknownMessageReceived(major, minor, params);
+                }
+            } else {
+                _listener.unknownMessageReceived(major, minor, params);
+            }
         } else if ("NAMING".equals(major)) {
             if ("REPLY".equals(minor)) {
                 String name = params.getProperty("NAME");

apps/sam/java/src/net/i2p/sam/client/SAMStreamSink.java

@@ -127,7 +127,7 @@ public class SAMStreamSink {
                 throw new IOException("handshake failed");
             if (_log.shouldLog(Log.DEBUG))
                 _log.debug("Handshake complete.  we are " + ourDest);
-            if (_isV3) {
+            if (_isV3 && mode != V1DG && mode != V1RAW) {
                 Socket sock2 = connect(isSSL);
                 out = sock2.getOutputStream();
                 eventHandler = new SinkEventHandler2(_context, sock2.getInputStream(), out);
@@ -158,7 +158,7 @@ public class SAMStreamSink {
 
         @Override
         public void streamClosedReceived(String result, String id, String message) {
-            Sink sink = null;
+            Sink sink;
             synchronized (_remotePeers) {
                 sink = _remotePeers.remove(id);
             }
@@ -173,7 +173,7 @@ public class SAMStreamSink {
 
         @Override
         public void streamDataReceived(String id, byte data[], int offset, int length) {
-            Sink sink = null;
+            Sink sink;
             synchronized (_remotePeers) {
                 sink = _remotePeers.get(id);
             }
@@ -212,6 +212,34 @@ public class SAMStreamSink {
                 }
             }
         }
+
+        @Override
+        public void datagramReceived(String dest, byte[] data, int offset, int length, int fromPort, int toPort) {
+            // just get the first
+            Sink sink;
+            synchronized (_remotePeers) {
+                if (_remotePeers.isEmpty()) {
+                    _log.error("not connected but we received datagram " + length + "?");
+                    return;
+                }
+                sink = _remotePeers.values().iterator().next();
+            }
+            sink.received(data, offset, length);
+        }
+
+        @Override
+        public void rawReceived(byte[] data, int offset, int length, int fromPort, int toPort, int protocol) {
+            // just get the first
+            Sink sink;
+            synchronized (_remotePeers) {
+                if (_remotePeers.isEmpty()) {
+                    _log.error("not connected but we received raw " + length + "?");
+                    return;
+                }
+                sink = _remotePeers.values().iterator().next();
+            }
+            sink.received(data, offset, length);
+        }
     }
 
     private class SinkEventHandler2 extends SinkEventHandler {
@@ -369,12 +397,13 @@ public class SAMStreamSink {
                 samOut.flush();
                 if (_log.shouldLog(Log.DEBUG))
                     _log.debug("Session create sent");
-                boolean ok = eventHandler.waitForSessionCreateReply();
+                if (mode == STREAM) {
-                if (!ok) 
+                    boolean ok = eventHandler.waitForSessionCreateReply();
-                    throw new IOException("Session create failed");
+                    if (!ok) 
-                if (_log.shouldLog(Log.DEBUG))
+                        throw new IOException("Session create failed");
-                    _log.debug("Session create reply found: " + ok);
+                    if (_log.shouldLog(Log.DEBUG))
-
+                        _log.debug("Session create reply found: " + ok);
+                }
                 req = "NAMING LOOKUP NAME=ME\n";
                 samOut.write(req.getBytes());
                 samOut.flush();
@@ -386,9 +415,12 @@ public class SAMStreamSink {
                 if (destination == null) {
                     _log.error("No naming lookup reply found!");
                     return null;
-                } else {
+                }
-                    if (_log.shouldInfo())
+                if (_log.shouldInfo())
-                        _log.info(_destFile + " is located at " + destination);
+                    _log.info(_destFile + " is located at " + destination);
+                if (mode != STREAM) {
+                    // fake it so the sink starts
+                    eventHandler.streamConnectedReceived(destination, "FAKE");
                 }
                 return destination;
             } catch (IOException e) {