jruby · Dec 24, 2015 · Dec 24, 2015 · Dec 26, 2015
Showing with 120 additions and 12 deletions.

+116 −8 core/src/main/java/org/jruby/ext/socket/RubyUNIXSocket.java

+4 −4 test/mri/excludes/TestSocket_UNIXSocket.rb
diff --git a/core/src/main/java/org/jruby/ext/socket/RubyUNIXSocket.java b/core/src/main/java/org/jruby/ext/socket/RubyUNIXSocket.java
@@ -36,10 +36,15 @@
 import jnr.unixsocket.UnixServerSocketChannel;
 import jnr.unixsocket.UnixSocketAddress;
 import jnr.unixsocket.UnixSocketChannel;
+import jnr.posix.CmsgHdr;
+import jnr.posix.MsgHdr;
+import jnr.posix.POSIX;
 import org.jruby.Ruby;
 import org.jruby.RubyClass;
 import org.jruby.RubyNumeric;
 import org.jruby.RubyString;
+import org.jruby.RubyIO;
+import org.jruby.RubyFixnum;
 import org.jruby.anno.JRubyClass;
 import org.jruby.anno.JRubyMethod;
 import org.jruby.runtime.Helpers;
@@ -49,10 +54,14 @@
 import org.jruby.runtime.builtin.IRubyObject;
 import org.jruby.util.ByteList;
 import org.jruby.util.io.ModeFlags;
+import org.jruby.util.io.OpenFile;
+import org.jruby.util.io.FilenoUtil;
 
 import java.io.File;
 import java.io.IOException;
 import java.nio.channels.Channel;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 
 
 @JRubyClass(name="UNIXSocket", parent="BasicSocket")
@@ -145,16 +154,115 @@ public IRubyObject recvfrom(ThreadContext context, IRubyObject[] args) {
                 peeraddr(context));
     }
 
-    @JRubyMethod(notImplemented = true)
-    public IRubyObject send_io(IRubyObject path) {
-        //TODO: implement, won't do this now
-        return  getRuntime().getNil();
+    @JRubyMethod
+    public IRubyObject send_io(ThreadContext context, IRubyObject arg) {
+        final Ruby runtime = context.runtime;
+        final POSIX posix = runtime.getPosix();
+        OpenFile fptr = getOpenFileChecked();
+        int fd;
+
+        if (arg.callMethod(context, "kind_of?", runtime.getIO()).isTrue()) {
+          fd = ((RubyIO) arg).getOpenFileChecked().getFileno();
+        } else if (arg.callMethod(context, "kind_of?", runtime.getFixnum()).isTrue()) {
+          fd = ((RubyFixnum) arg).getIntValue();
+        } else {
+          throw runtime.newTypeError("neither IO nor file descriptor");
+        }
+
+        if (FilenoUtil.isFake(fd)) {
+          throw runtime.newTypeError("file descriptor is not native");
+        }
+
+        byte[] dataBytes = new byte[1];
+        dataBytes[0] = 0;
+
+        MsgHdr outMessage = posix.allocateMsgHdr();
+
+        ByteBuffer[] outIov = new ByteBuffer[1];
+        outIov[0] = ByteBuffer.allocateDirect(dataBytes.length);
+        outIov[0].put(dataBytes);
+        outIov[0].flip();
+
+        outMessage.setIov(outIov);
+
+        CmsgHdr outControl = outMessage.allocateControl(4);
+        outControl.setLevel(SocketLevel.SOL_SOCKET.intValue());
+        outControl.setType(0x01);
+
+        ByteBuffer fdBuf = ByteBuffer.allocateDirect(4);
+        fdBuf.order(ByteOrder.nativeOrder());
+        fdBuf.putInt(0, fd);
+        outControl.setData(fdBuf);
+
+        boolean locked = fptr.lock();
+        try {
+            while (posix.sendmsg(fptr.getFileno(), outMessage, 0) == -1) {
+                if (!fptr.waitWritable(context)) {
+                    throw runtime.newErrnoFromInt(posix.errno(), "sendmsg(2)");
+                }
+            }
+        } finally {
+            if (locked) fptr.unlock();
+        }
+
+        return runtime.getNil();
     }
 
-    @JRubyMethod(rest = true, notImplemented = true)
-    public IRubyObject recv_io(IRubyObject[] args) {
-        //TODO: implement, won't do this now
-        return  getRuntime().getNil();
+    @JRubyMethod(optional = 2)
+    public IRubyObject recv_io(ThreadContext context, IRubyObject[] args) {
+        final Ruby runtime = context.runtime;
+        final POSIX posix = runtime.getPosix();
+        OpenFile fptr = getOpenFileChecked();
+
+        IRubyObject klass = runtime.getIO();
+        IRubyObject mode = runtime.getNil();
+        if (args.length > 0) {
+            klass = args[0];
+        }
+        if (args.length > 1) {
+            mode = args[1];
+        }
+
+        MsgHdr inMessage = posix.allocateMsgHdr();
+        ByteBuffer[] inIov = new ByteBuffer[1];
+        inIov[0] = ByteBuffer.allocateDirect(1);
+        inMessage.setIov(inIov);
+
+        CmsgHdr inControl = inMessage.allocateControl(4);
+        inControl.setLevel(SocketLevel.SOL_SOCKET.intValue());
+        inControl.setType(0x01);
+
+        ByteBuffer fdBuf = ByteBuffer.allocateDirect(4);
+        fdBuf.order(ByteOrder.nativeOrder());
+        fdBuf.putInt(0, -1);
+        inControl.setData(fdBuf);
+
+        boolean locked = fptr.lock();
+        try {
+            while (posix.recvmsg(fptr.getFileno(), inMessage, 0) == -1) {
+                if (!fptr.waitReadable(context)) {
+                    throw runtime.newErrnoFromInt(posix.errno(), "recvmsg(2)");
+                }
+            }
+        } finally {
+            if (locked) fptr.unlock();
+        }
+
+
+        ByteBuffer inFdBuf = inMessage.getControls()[0].getData();
+        inFdBuf.order(ByteOrder.nativeOrder());
+
+        IRubyObject fd = runtime.newFixnum(inFdBuf.getInt());
+
+        if (klass.isNil()) {
+            return fd;
+        } else {
+            if (mode.isNil()) {
+              return Helpers.invoke(context, klass, "for_fd", fd);
+            } else {
+              return Helpers.invoke(context, klass, "for_fd", fd, mode);
+            }
+        }
     }
 
     @JRubyMethod(name = {"socketpair", "pair"}, optional = 2, meta = true)

diff --git a/test/mri/excludes/TestSocket_UNIXSocket.rb b/test/mri/excludes/TestSocket_UNIXSocket.rb
@@ -1,13 +1,13 @@
 exclude :test_addr, "needs investigation"
 exclude :test_dgram_pair, "needs investigation"
-exclude :test_epipe, "needs investigation"
+exclude :test_fd_passing, "close_on_exec? not implemented"
 exclude :test_fd_passing_race_condition, "needs investigation"
 exclude :test_getpeereid, "needs investigation"
 exclude :test_initialize, "needs investigation"
-exclude :test_noname_recv_nonblock, "needs investigation"
 exclude :test_noname_unpack_sockaddr_un, "needs investigation"
-exclude :test_nul, "needs investigation"
 exclude :test_socket_pair_with_block, "needs investigation"
 exclude :test_too_long_path, "needs investigation"
 exclude :test_unix_server_socket, "needs investigation"
-exclude :test_unix_socket_pair_with_block, "needs investigation"
+exclude :test_unix_socket_pair_with_block, "needs investigation"
+exclude :test_cloexec, "needs investigation"
+exclude :test_dgram_pair_sendrecvmsg_errno_set, "needs investigation"