jruby · Sep 17, 2015 · Sep 17, 2015 · Sep 17, 2015
diff --git a/core/src/main/java/org/jruby/RubyIO.java b/core/src/main/java/org/jruby/RubyIO.java
@@ -67,42 +67,41 @@
 import java.nio.channels.WritableByteChannel;
 import java.util.Arrays;
 import java.util.HashSet;
-import java.util.Map;
 import java.util.Set;
 
 import org.jcodings.Encoding;
+import org.jruby.ast.util.ArgsUtil;
 import org.jruby.anno.FrameField;
 import org.jruby.anno.JRubyMethod;
 import org.jruby.anno.JRubyClass;
 import org.jruby.common.IRubyWarnings.ID;
 import org.jruby.exceptions.RaiseException;
 import org.jruby.ext.fcntl.FcntlLibrary;
+import org.jruby.internal.runtime.ThreadedRunnable;
 import org.jruby.platform.Platform;
+import org.jruby.runtime.Arity;
 import org.jruby.runtime.Block;
 import org.jruby.runtime.ClassIndex;
 import org.jruby.runtime.ObjectAllocator;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.Visibility;
-import static org.jruby.runtime.Visibility.*;
 import org.jruby.runtime.builtin.IRubyObject;
+import org.jruby.runtime.encoding.EncodingService;
 import org.jruby.util.ByteList;
-import org.jruby.util.io.SelectExecutor;
-import org.jruby.util.io.IOOptions;
+import org.jruby.util.ShellLauncher.POpenProcess;
 import org.jruby.util.SafePropertyAccessor;
 import org.jruby.util.ShellLauncher;
 import org.jruby.util.TypeConverter;
+import org.jruby.util.io.IOEncodable;
+import org.jruby.util.io.IOOptions;
 import org.jruby.util.io.InvalidValueException;
-import org.jruby.util.io.STDIO;
 import org.jruby.util.io.OpenFile;
-
-import org.jruby.runtime.Arity;
+import org.jruby.util.io.SelectExecutor;
+import org.jruby.util.io.STDIO;
 
 import static org.jruby.RubyEnumerator.enumeratorize;
-import org.jruby.ast.util.ArgsUtil;
-import org.jruby.internal.runtime.ThreadedRunnable;
-import org.jruby.runtime.encoding.EncodingService;
-import org.jruby.util.ShellLauncher.POpenProcess;
-import org.jruby.util.io.IOEncodable;
+import static org.jruby.runtime.Visibility.*;
+import static org.jruby.util.io.ChannelHelper.*;
 
 /**
  *
@@ -134,7 +133,7 @@ public RubyIO(Ruby runtime, OutputStream outputStream, boolean autoclose) {
         }
 
         openFile = MakeOpenFile();
-        openFile.setFD(new ChannelFD(Channels.newChannel(outputStream), runtime.getPosix(), runtime.getFilenoUtil()));
+        openFile.setFD(new ChannelFD(writableChannel(outputStream), runtime.getPosix(), runtime.getFilenoUtil()));
         openFile.setMode(OpenFile.WRITABLE | OpenFile.APPEND);
         openFile.setAutoclose(autoclose);
     }
@@ -147,7 +146,7 @@ public RubyIO(Ruby runtime, InputStream inputStream) {
         }
 
         openFile = MakeOpenFile();
-        openFile.setFD(new ChannelFD(Channels.newChannel(inputStream), runtime.getPosix(), runtime.getFilenoUtil()));
+        openFile.setFD(new ChannelFD(readableChannel(inputStream), runtime.getPosix(), runtime.getFilenoUtil()));
         openFile.setMode(OpenFile.READABLE);
     }
 

diff --git a/core/src/main/java/org/jruby/util/io/ChannelFD.java b/core/src/main/java/org/jruby/util/io/ChannelFD.java
@@ -14,6 +14,7 @@
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SocketChannel;
 import java.nio.channels.WritableByteChannel;
@@ -130,7 +131,7 @@ private void initChannelTypes() {
         else chRead = null;
         if (ch instanceof WritableByteChannel) chWrite = (WritableByteChannel)ch;
         else chWrite = null;
-        if (ch instanceof FileChannel) chSeek = (FileChannel)ch;
+        if (ch instanceof SeekableByteChannel) chSeek = (SeekableByteChannel)ch;
         else chSeek = null;
         if (ch instanceof SelectableChannel) chSelect = (SelectableChannel)ch;
         else chSelect = null;
@@ -154,7 +155,7 @@ private void initChannelTypes() {
     public Channel ch;
     public ReadableByteChannel chRead;
     public WritableByteChannel chWrite;
-    public FileChannel chSeek;
+    public SeekableByteChannel chSeek;
     public SelectableChannel chSelect;
     public FileChannel chFile;
     public SocketChannel chSock;
@@ -163,7 +164,7 @@ private void initChannelTypes() {
     public int fakeFileno;
     private AtomicInteger refs;
     public FileLock currentLock;
-    private POSIX posix;
+    private final POSIX posix;
     public boolean isNativeFile = false;
     private final FilenoUtil filenoUtil;
 }
diff --git a/core/src/main/java/org/jruby/util/io/ChannelHelper.java b/core/src/main/java/org/jruby/util/io/ChannelHelper.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2015 JRuby.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    JRuby - initial API and implementation and/or initial documentation
+ */
+package org.jruby.util.io;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
+
+/**
+ * Helper that attempts to improve Channels' static helpers.
+ * @author kares
+ */
+public abstract class ChannelHelper {
+
+    private ChannelHelper() { /* */ }
+
+    public static ReadableByteChannel readableChannel(final InputStream inputStream) {
+        if (inputStream instanceof ByteArrayInputStream) {
+            return new SeekableByteChannelImpl((ByteArrayInputStream) inputStream);
+        }
+        return Channels.newChannel(inputStream);
+    }
+
+    public static WritableByteChannel writableChannel(final OutputStream ouputStream) {
+        return Channels.newChannel(ouputStream);
+    }
+
+}
diff --git a/core/src/main/java/org/jruby/util/io/OpenFile.java b/core/src/main/java/org/jruby/util/io/OpenFile.java
@@ -34,6 +34,7 @@
 import java.nio.channels.FileChannel;
 import java.nio.channels.FileLock;
 import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
@@ -71,7 +72,7 @@ public OpenFile(IRubyObject nil) {
     public static final int TEXTMODE           = 0x00001000;
     public static final int SETENC_BY_BOM      = 0x00100000;
     public static final int PREP         = (1<<16);
-    
+
     public static final int SYNCWRITE = SYNC | WRITABLE;
 
     public static final int PIPE_BUF = 512; // value of _POSIX_PIPE_BUF from Mac OS X 10.9
@@ -193,7 +194,7 @@ public void setChannel(Channel fd) {
     public int getMode() {
         return mode;
     }
-    
+
     public String getModeAsString(Ruby runtime) {
         String modeString = getStringFromMode(mode);
 
@@ -203,10 +204,10 @@ public String getModeAsString(Ruby runtime) {
 
         return modeString;
     }
-    
+
     public static int getModeFlagsAsIntFrom(int fmode) {
         int oflags = 0;
-        
+
         if ((fmode & READABLE) != 0) {
             if ((fmode & WRITABLE) != 0) {
                 oflags |= ModeFlags.RDWR;
@@ -216,13 +217,13 @@ public static int getModeFlagsAsIntFrom(int fmode) {
         } else if ((fmode & WRITABLE) != 0) {
             oflags |= ModeFlags.WRONLY;
         }
-        
+
         if ((fmode & APPEND) != 0) oflags |= ModeFlags.APPEND;
         if ((fmode & CREATE) != 0) oflags |= ModeFlags.CREAT;
         if ((fmode & BINMODE) != 0) oflags |= ModeFlags.BINARY;
         if ((fmode & TEXTMODE) != 0) oflags |= ModeFlags.TEXT;
         if ((fmode & TRUNC) != 0) oflags |= ModeFlags.TRUNC;
-        
+
         return oflags;
     }
 
@@ -2257,7 +2258,7 @@ public WritableByteChannel writeChannel() {
         return fd.chWrite;
     }
 
-    public FileChannel seekChannel() {
+    public SeekableByteChannel seekChannel() {
         return fd.chSeek;
     }
 

diff --git a/core/src/main/java/org/jruby/util/io/SeekableByteChannelImpl.java b/core/src/main/java/org/jruby/util/io/SeekableByteChannelImpl.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright (c) 2015 JRuby.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    JRuby - initial API and implementation and/or initial documentation
+ */
+package org.jruby.util.io;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.channels.spi.AbstractInterruptibleChannel;
+
+/**
+ *
+ * @author kares
+ */
+final class SeekableByteChannelImpl extends AbstractInterruptibleChannel  // Not really interruptible
+    implements ReadableByteChannel, SeekableByteChannel {
+
+    private static final byte[] EMPTY_BUF = new byte[0];
+
+    private final ByteArrayInputStream in;
+    private byte buf[] = EMPTY_BUF;
+    private final Object readLock = new Object();
+
+    private final int mark;
+    private final int count;
+
+    private int truncatedBy = 0;
+
+    SeekableByteChannelImpl(ByteArrayInputStream in) {
+        this.in = in;
+        this.mark = ByteArrayInputStreamHelper.mark(in);
+        this.count = ByteArrayInputStreamHelper.count(in);
+    }
+
+    @Override // based on java.nio.channels.Channels.ReadableByteChannelImpl
+    public int read(ByteBuffer dst) throws IOException {
+        final int len = dst.remaining();
+        int totalRead = 0; int bytesRead = 0;
+        synchronized (readLock) {
+            while (totalRead < len) {
+                int bytesToRead = Math.min((len - totalRead), 4096);
+                if (buf.length < bytesToRead) buf = new byte[bytesToRead];
+                final int avail = in.available() - truncatedBy;
+                if ( totalRead > 0 && avail <= 0 ) break; // block at most once
+                try {
+                    begin();
+                    // make sure we account for truncated bytes :
+                    if (bytesToRead > avail) bytesToRead = avail;
+
+                    bytesRead = in.read(buf, 0, bytesToRead);
+                } finally {
+                    end(bytesRead > 0);
+                }
+                if (bytesRead < 0) break;
+                else totalRead += bytesRead;
+                dst.put(buf, 0, bytesRead);
+            }
+            if ((bytesRead < 0) && (totalRead == 0)) return -1;
+
+            return totalRead;
+        }
+    }
+
+    @Override
+    protected void implCloseChannel() throws IOException { in.close(); }
+
+    // SeekableByteChannel interface :
+
+    public long position() {
+        return ByteArrayInputStreamHelper.pos(in) - mark;
+    }
+
+    public SeekableByteChannel position(long newPosition) throws IOException {
+        if ( newPosition < 0 ) {
+            throw new IllegalArgumentException("negative new position: " + newPosition);
+        }
+        if ( newPosition > Integer.MAX_VALUE ) {
+            throw new IllegalArgumentException("can not set new position: " + newPosition + " too big!");
+        }
+        synchronized (readLock) {
+            this.in.reset(); // to initial mark (0 or offset)
+            if ( newPosition > 0 ) this.in.skip(newPosition);
+            //this.position = (int) newPosition;
+        }
+        return this;
+    }
+
+    public long size() {
+        return Math.max(count - truncatedBy, 0);
+    }
+
+    public SeekableByteChannel truncate(long size) throws IOException {
+        if ( size < 0 ) {
+            throw new IllegalArgumentException("negative truncate size given: " + size);
+        }
+        final int s = Math.min((int) size(), size > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int) size);
+        this.truncatedBy += s;
+        return this;
+    }
+
+    public int write(ByteBuffer src) throws IOException {
+        throw new UnsupportedOperationException("write not supported");
+    }
+
+    private static class ByteArrayInputStreamHelper {
+
+        static int pos(ByteArrayInputStream in) {
+            return readField(in, "pos"); // current pos
+        }
+
+        static int count(ByteArrayInputStream in) {
+            return readField(in, "count"); // buf.length or offset + length
+        }
+
+        static int mark(ByteArrayInputStream in) {
+            return readField(in, "mark"); // 0 oir offset or if marked previously
+        }
+
+        private static int readField(ByteArrayInputStream self, String name) {
+            try {
+                Field field = ByteArrayInputStream.class.getDeclaredField(name);
+                field.setAccessible(true);
+                return field.getInt(self);
+            }
+            catch (NoSuchFieldException ex) { throw new RuntimeException(ex); } // should never happen
+            catch (IllegalAccessException ex) {
+                throw new IllegalStateException(ex); // TODO
+            }
+        }
+
+    }
+
+}