src/java.net.http/share/classes/jdk/internal/net/http/RequestPublishers.java - platform/libcore - Git at Google

 /*
  * Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License version 2 only, as
  * published by the Free Software Foundation.  Oracle designates this
  * particular file as subject to the "Classpath" exception as provided
  * by Oracle in the LICENSE file that accompanied this code.
  *
  * This code is distributed in the hope that it will be useful, but WITHOUT
  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  * version 2 for more details (a copy is included in the LICENSE file that
  * accompanied this code).
  *
  * You should have received a copy of the GNU General Public License version
  * 2 along with this work; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  *
  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  * or visit www.oracle.com if you need additional information or have any
  * questions.
  */

 package jdk.internal.net.http;

 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FilePermission;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.AccessControlContext;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.security.PrivilegedActionException;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Flow;
 import java.util.concurrent.Flow.Publisher;
 import java.util.function.Supplier;
 import java.net.http.HttpRequest.BodyPublisher;
 import jdk.internal.net.http.common.Utils;

 public final class RequestPublishers {

     private RequestPublishers() { }

     public static class ByteArrayPublisher implements BodyPublisher {
         private final int length;
         private final byte[] content;
         private final int offset;
         private final int bufSize;

         public ByteArrayPublisher(byte[] content) {
             this(content, 0, content.length);
         }

         public ByteArrayPublisher(byte[] content, int offset, int length) {
             this(content, offset, length, Utils.BUFSIZE);
         }

         /* bufSize exposed for testing purposes */
         ByteArrayPublisher(byte[] content, int offset, int length, int bufSize) {
             this.content = content;
             this.offset = offset;
             this.length = length;
             this.bufSize = bufSize;
         }

         List<ByteBuffer> copy(byte[] content, int offset, int length) {
             List<ByteBuffer> bufs = new ArrayList<>();
             while (length > 0) {
                 ByteBuffer b = ByteBuffer.allocate(Math.min(bufSize, length));
                 int max = b.capacity();
                 int tocopy = Math.min(max, length);
                 b.put(content, offset, tocopy);
                 offset += tocopy;
                 length -= tocopy;
                 b.flip();
                 bufs.add(b);
             }
             return bufs;
         }

         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             List<ByteBuffer> copy = copy(content, offset, length);
             var delegate = new PullPublisher<>(copy);
             delegate.subscribe(subscriber);
         }

         @Override
         public long contentLength() {
             return length;
         }
     }

     // This implementation has lots of room for improvement.
     public static class IterablePublisher implements BodyPublisher {
         private final Iterable<byte[]> content;
         private volatile long contentLength;

         public IterablePublisher(Iterable<byte[]> content) {
             this.content = Objects.requireNonNull(content);
         }

         // The ByteBufferIterator will iterate over the byte[] arrays in
         // the content one at the time.
         //
         class ByteBufferIterator implements Iterator<ByteBuffer> {
             final ConcurrentLinkedQueue<ByteBuffer> buffers = new ConcurrentLinkedQueue<>();
             final Iterator<byte[]> iterator = content.iterator();
             @Override
             public boolean hasNext() {
                 return !buffers.isEmpty() || iterator.hasNext();
             }

             @Override
             public ByteBuffer next() {
                 ByteBuffer buffer = buffers.poll();
                 while (buffer == null) {
                     copy();
                     buffer = buffers.poll();
                 }
                 return buffer;
             }

             ByteBuffer getBuffer() {
                 return Utils.getBuffer();
             }

             void copy() {
                 byte[] bytes = iterator.next();
                 int length = bytes.length;
                 if (length == 0 && iterator.hasNext()) {
                     // avoid inserting empty buffers, except
                     // if that's the last.
                     return;
                 }
                 int offset = 0;
                 do {
                     ByteBuffer b = getBuffer();
                     int max = b.capacity();

                     int tocopy = Math.min(max, length);
                     b.put(bytes, offset, tocopy);
                     offset += tocopy;
                     length -= tocopy;
                     b.flip();
                     buffers.add(b);
                 } while (length > 0);
             }
         }

         public Iterator<ByteBuffer> iterator() {
             return new ByteBufferIterator();
         }

         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             Iterable<ByteBuffer> iterable = this::iterator;
             var delegate = new PullPublisher<>(iterable);
             delegate.subscribe(subscriber);
         }

         static long computeLength(Iterable<byte[]> bytes) {
             long len = 0;
             for (byte[] b : bytes) {
                 len = Math.addExact(len, (long)b.length);
             }
             return len;
         }

         @Override
         public long contentLength() {
             if (contentLength == 0) {
                 synchronized(this) {
                     if (contentLength == 0) {
                         contentLength = computeLength(content);
                     }
                 }
             }
             return contentLength;
         }
     }

     public static class StringPublisher extends ByteArrayPublisher {
         public StringPublisher(String content, Charset charset) {
             super(content.getBytes(charset));
         }
     }

     public static class EmptyPublisher implements BodyPublisher {
         private final Flow.Publisher<ByteBuffer> delegate =
                 new PullPublisher<ByteBuffer>(Collections.emptyList(), null);

         @Override
         public long contentLength() {
             return 0;
         }

         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             delegate.subscribe(subscriber);
         }
     }

     /**
      * Publishes the content of a given file.
      *
      * Privileged actions are performed within a limited doPrivileged that only
      * asserts the specific, read, file permission that was checked during the
      * construction of this FilePublisher.
      */
     public static class FilePublisher implements BodyPublisher  {

         private static final FilePermission[] EMPTY_FILE_PERMISSIONS = new FilePermission[0];

         private final File file;
         private final FilePermission[] filePermissions;

         private static String pathForSecurityCheck(Path path) {
             return path.toFile().getPath();
         }

         /**
          * Factory for creating FilePublisher.
          *
          * Permission checks are performed here before construction of the
          * FilePublisher. Permission checking and construction are deliberately
          * and tightly co-located.
          */
         public static FilePublisher create(Path path) throws FileNotFoundException {
             FilePermission filePermission = null;
             SecurityManager sm = System.getSecurityManager();
             if (sm != null) {
                 String fn = pathForSecurityCheck(path);
                 FilePermission readPermission = new FilePermission(fn, "read");
                 sm.checkPermission(readPermission);
                 filePermission = readPermission;
             }

             // existence check must be after permission checks
             if (Files.notExists(path))
                 throw new FileNotFoundException(path + " not found");

             return new FilePublisher(path, filePermission);
         }

         private FilePublisher(Path name, FilePermission filePermission) {
             assert filePermission != null ? filePermission.getActions().equals("read") : true;
             file = name.toFile();
             this.filePermissions = filePermission == null ? EMPTY_FILE_PERMISSIONS
                     : new FilePermission[] { filePermission };
         }

         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             InputStream is;
             if (System.getSecurityManager() == null) {
                 try {
                     is = new FileInputStream(file);
                 } catch (IOException ioe) {
                     throw new UncheckedIOException(ioe);
                 }
             } else {
                 try {
                     PrivilegedExceptionAction<FileInputStream> pa =
                             () -> new FileInputStream(file);
                     is = AccessController.doPrivileged(pa, null, filePermissions);
                 } catch (PrivilegedActionException pae) {
                     throw new UncheckedIOException((IOException) pae.getCause());
                 }
             }
             PullPublisher<ByteBuffer> publisher =
                     new PullPublisher<>(() -> new StreamIterator(is));
             publisher.subscribe(subscriber);
         }

         @Override
         public long contentLength() {
             if (System.getSecurityManager() == null) {
                 return file.length();
             } else {
                 PrivilegedAction<Long> pa = () -> file.length();
                 return AccessController.doPrivileged(pa, null, filePermissions);
             }
         }
     }

     /**
      * Reads one buffer ahead all the time, blocking in hasNext()
      */
     public static class StreamIterator implements Iterator<ByteBuffer> {
         final InputStream is;
         final Supplier<? extends ByteBuffer> bufSupplier;
         volatile ByteBuffer nextBuffer;
         volatile boolean need2Read = true;
         volatile boolean haveNext;

         StreamIterator(InputStream is) {
             this(is, Utils::getBuffer);
         }

         StreamIterator(InputStream is, Supplier<? extends ByteBuffer> bufSupplier) {
             this.is = is;
             this.bufSupplier = bufSupplier;
         }

 //        Throwable error() {
 //            return error;
 //        }

         private int read() {
             nextBuffer = bufSupplier.get();
             nextBuffer.clear();
             byte[] buf = nextBuffer.array();
             int offset = nextBuffer.arrayOffset();
             int cap = nextBuffer.capacity();
             try {
                 int n = is.read(buf, offset, cap);
                 if (n == -1) {
                     is.close();
                     return -1;
                 }
                 //flip
                 nextBuffer.limit(n);
                 nextBuffer.position(0);
                 return n;
             } catch (IOException ex) {
                 return -1;
             }
         }

         @Override
         public synchronized boolean hasNext() {
             if (need2Read) {
                 haveNext = read() != -1;
                 if (haveNext) {
                     need2Read = false;
                 }
                 return haveNext;
             }
             return haveNext;
         }

         @Override
         public synchronized ByteBuffer next() {
             if (!hasNext()) {
                 throw new NoSuchElementException();
             }
             need2Read = true;
             return nextBuffer;
         }

     }

     public static class InputStreamPublisher implements BodyPublisher {
         private final Supplier<? extends InputStream> streamSupplier;

         public InputStreamPublisher(Supplier<? extends InputStream> streamSupplier) {
             this.streamSupplier = Objects.requireNonNull(streamSupplier);
         }

         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             PullPublisher<ByteBuffer> publisher;
             InputStream is = streamSupplier.get();
             if (is == null) {
                 Throwable t = new IOException("streamSupplier returned null");
                 publisher = new PullPublisher<>(null, t);
             } else  {
                 publisher = new PullPublisher<>(iterableOf(is), null);
             }
             publisher.subscribe(subscriber);
         }

         protected Iterable<ByteBuffer> iterableOf(InputStream is) {
             return () -> new StreamIterator(is);
         }

         @Override
         public long contentLength() {
             return -1;
         }
     }

     public static final class PublisherAdapter implements BodyPublisher {

         private final Publisher<? extends ByteBuffer> publisher;
         private final long contentLength;

         public PublisherAdapter(Publisher<? extends ByteBuffer> publisher,
                          long contentLength) {
             this.publisher = Objects.requireNonNull(publisher);
             this.contentLength = contentLength;
         }

         @Override
         public final long contentLength() {
             return contentLength;
         }

         @Override
         public final void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             publisher.subscribe(subscriber);
         }
     }
 }
	/*
	* Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
	* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
	*
	* This code is free software; you can redistribute it and/or modify it
	* under the terms of the GNU General Public License version 2 only, as
	* published by the Free Software Foundation. Oracle designates this
	* particular file as subject to the "Classpath" exception as provided
	* by Oracle in the LICENSE file that accompanied this code.
	*
	* This code is distributed in the hope that it will be useful, but WITHOUT
	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	* version 2 for more details (a copy is included in the LICENSE file that
	* accompanied this code).
	*
	* You should have received a copy of the GNU General Public License version
	* 2 along with this work; if not, write to the Free Software Foundation,
	* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
	*
	* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
	* or visit www.oracle.com if you need additional information or have any
	* questions.
	*/

	package jdk.internal.net.http;

	import java.io.File;
	import java.io.FileInputStream;
	import java.io.FileNotFoundException;
	import java.io.FilePermission;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.UncheckedIOException;
	import java.nio.ByteBuffer;
	import java.nio.charset.Charset;
	import java.nio.file.Files;
	import java.nio.file.Path;
	import java.security.AccessControlContext;
	import java.security.AccessController;
	import java.security.PrivilegedAction;
	import java.security.PrivilegedActionException;
	import java.security.PrivilegedExceptionAction;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Iterator;
	import java.util.List;
	import java.util.NoSuchElementException;
	import java.util.Objects;
	import java.util.concurrent.ConcurrentLinkedQueue;
	import java.util.concurrent.Flow;
	import java.util.concurrent.Flow.Publisher;
	import java.util.function.Supplier;
	import java.net.http.HttpRequest.BodyPublisher;
	import jdk.internal.net.http.common.Utils;

	public final class RequestPublishers {

	private RequestPublishers() { }

	public static class ByteArrayPublisher implements BodyPublisher {
	private final int length;
	private final byte[] content;
	private final int offset;
	private final int bufSize;

	public ByteArrayPublisher(byte[] content) {
	this(content, 0, content.length);
	}

	public ByteArrayPublisher(byte[] content, int offset, int length) {
	this(content, offset, length, Utils.BUFSIZE);
	}

	/* bufSize exposed for testing purposes */
	ByteArrayPublisher(byte[] content, int offset, int length, int bufSize) {
	this.content = content;
	this.offset = offset;
	this.length = length;
	this.bufSize = bufSize;
	}

	List<ByteBuffer> copy(byte[] content, int offset, int length) {
	List<ByteBuffer> bufs = new ArrayList<>();
	while (length > 0) {
	ByteBuffer b = ByteBuffer.allocate(Math.min(bufSize, length));
	int max = b.capacity();
	int tocopy = Math.min(max, length);
	b.put(content, offset, tocopy);
	offset += tocopy;
	length -= tocopy;
	b.flip();
	bufs.add(b);
	}
	return bufs;
	}

	@Override
	public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	List<ByteBuffer> copy = copy(content, offset, length);
	var delegate = new PullPublisher<>(copy);
	delegate.subscribe(subscriber);
	}

	@Override
	public long contentLength() {
	return length;
	}
	}

	// This implementation has lots of room for improvement.
	public static class IterablePublisher implements BodyPublisher {
	private final Iterable<byte[]> content;
	private volatile long contentLength;

	public IterablePublisher(Iterable<byte[]> content) {
	this.content = Objects.requireNonNull(content);
	}

	// The ByteBufferIterator will iterate over the byte[] arrays in
	// the content one at the time.
	//
	class ByteBufferIterator implements Iterator<ByteBuffer> {
	final ConcurrentLinkedQueue<ByteBuffer> buffers = new ConcurrentLinkedQueue<>();
	final Iterator<byte[]> iterator = content.iterator();
	@Override
	public boolean hasNext() {
	return !buffers.isEmpty() \|\| iterator.hasNext();
	}

	@Override
	public ByteBuffer next() {
	ByteBuffer buffer = buffers.poll();
	while (buffer == null) {
	copy();
	buffer = buffers.poll();
	}
	return buffer;
	}

	ByteBuffer getBuffer() {
	return Utils.getBuffer();
	}

	void copy() {
	byte[] bytes = iterator.next();
	int length = bytes.length;
	if (length == 0 && iterator.hasNext()) {
	// avoid inserting empty buffers, except
	// if that's the last.
	return;
	}
	int offset = 0;
	do {
	ByteBuffer b = getBuffer();
	int max = b.capacity();

	int tocopy = Math.min(max, length);
	b.put(bytes, offset, tocopy);
	offset += tocopy;
	length -= tocopy;
	b.flip();
	buffers.add(b);
	} while (length > 0);
	}
	}

	public Iterator<ByteBuffer> iterator() {
	return new ByteBufferIterator();
	}

	@Override
	public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	Iterable<ByteBuffer> iterable = this::iterator;
	var delegate = new PullPublisher<>(iterable);
	delegate.subscribe(subscriber);
	}

	static long computeLength(Iterable<byte[]> bytes) {
	long len = 0;
	for (byte[] b : bytes) {
	len = Math.addExact(len, (long)b.length);
	}
	return len;
	}

	@Override
	public long contentLength() {
	if (contentLength == 0) {
	synchronized(this) {
	if (contentLength == 0) {
	contentLength = computeLength(content);
	}
	}
	}
	return contentLength;
	}
	}

	public static class StringPublisher extends ByteArrayPublisher {
	public StringPublisher(String content, Charset charset) {
	super(content.getBytes(charset));
	}
	}

	public static class EmptyPublisher implements BodyPublisher {
	private final Flow.Publisher<ByteBuffer> delegate =
	new PullPublisher<ByteBuffer>(Collections.emptyList(), null);

	@Override
	public long contentLength() {
	return 0;
	}

	@Override
	public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	delegate.subscribe(subscriber);
	}
	}

	/**
	* Publishes the content of a given file.
	*
	* Privileged actions are performed within a limited doPrivileged that only
	* asserts the specific, read, file permission that was checked during the
	* construction of this FilePublisher.
	*/
	public static class FilePublisher implements BodyPublisher {

	private static final FilePermission[] EMPTY_FILE_PERMISSIONS = new FilePermission[0];

	private final File file;
	private final FilePermission[] filePermissions;

	private static String pathForSecurityCheck(Path path) {
	return path.toFile().getPath();
	}

	/**
	* Factory for creating FilePublisher.
	*
	* Permission checks are performed here before construction of the
	* FilePublisher. Permission checking and construction are deliberately
	* and tightly co-located.
	*/
	public static FilePublisher create(Path path) throws FileNotFoundException {
	FilePermission filePermission = null;
	SecurityManager sm = System.getSecurityManager();
	if (sm != null) {
	String fn = pathForSecurityCheck(path);
	FilePermission readPermission = new FilePermission(fn, "read");
	sm.checkPermission(readPermission);
	filePermission = readPermission;
	}

	// existence check must be after permission checks
	if (Files.notExists(path))
	throw new FileNotFoundException(path + " not found");

	return new FilePublisher(path, filePermission);
	}

	private FilePublisher(Path name, FilePermission filePermission) {
	assert filePermission != null ? filePermission.getActions().equals("read") : true;
	file = name.toFile();
	this.filePermissions = filePermission == null ? EMPTY_FILE_PERMISSIONS
	: new FilePermission[] { filePermission };
	}

	@Override
	public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	InputStream is;
	if (System.getSecurityManager() == null) {
	try {
	is = new FileInputStream(file);
	} catch (IOException ioe) {
	throw new UncheckedIOException(ioe);
	}
	} else {
	try {
	PrivilegedExceptionAction<FileInputStream> pa =
	() -> new FileInputStream(file);
	is = AccessController.doPrivileged(pa, null, filePermissions);
	} catch (PrivilegedActionException pae) {
	throw new UncheckedIOException((IOException) pae.getCause());
	}
	}
	PullPublisher<ByteBuffer> publisher =
	new PullPublisher<>(() -> new StreamIterator(is));
	publisher.subscribe(subscriber);
	}

	@Override
	public long contentLength() {
	if (System.getSecurityManager() == null) {
	return file.length();
	} else {
	PrivilegedAction<Long> pa = () -> file.length();
	return AccessController.doPrivileged(pa, null, filePermissions);
	}
	}
	}

	/**
	* Reads one buffer ahead all the time, blocking in hasNext()
	*/
	public static class StreamIterator implements Iterator<ByteBuffer> {
	final InputStream is;
	final Supplier<? extends ByteBuffer> bufSupplier;
	volatile ByteBuffer nextBuffer;
	volatile boolean need2Read = true;
	volatile boolean haveNext;

	StreamIterator(InputStream is) {
	this(is, Utils::getBuffer);
	}

	StreamIterator(InputStream is, Supplier<? extends ByteBuffer> bufSupplier) {
	this.is = is;
	this.bufSupplier = bufSupplier;
	}

	// Throwable error() {
	// return error;
	// }

	private int read() {
	nextBuffer = bufSupplier.get();
	nextBuffer.clear();
	byte[] buf = nextBuffer.array();
	int offset = nextBuffer.arrayOffset();
	int cap = nextBuffer.capacity();
	try {
	int n = is.read(buf, offset, cap);
	if (n == -1) {
	is.close();
	return -1;
	}
	//flip
	nextBuffer.limit(n);
	nextBuffer.position(0);
	return n;
	} catch (IOException ex) {
	return -1;
	}
	}

	@Override
	public synchronized boolean hasNext() {
	if (need2Read) {
	haveNext = read() != -1;
	if (haveNext) {
	need2Read = false;
	}
	return haveNext;
	}
	return haveNext;
	}

	@Override
	public synchronized ByteBuffer next() {
	if (!hasNext()) {
	throw new NoSuchElementException();
	}
	need2Read = true;
	return nextBuffer;
	}

	}

	public static class InputStreamPublisher implements BodyPublisher {
	private final Supplier<? extends InputStream> streamSupplier;

	public InputStreamPublisher(Supplier<? extends InputStream> streamSupplier) {
	this.streamSupplier = Objects.requireNonNull(streamSupplier);
	}

	@Override
	public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	PullPublisher<ByteBuffer> publisher;
	InputStream is = streamSupplier.get();
	if (is == null) {
	Throwable t = new IOException("streamSupplier returned null");
	publisher = new PullPublisher<>(null, t);
	} else {
	publisher = new PullPublisher<>(iterableOf(is), null);
	}
	publisher.subscribe(subscriber);
	}

	protected Iterable<ByteBuffer> iterableOf(InputStream is) {
	return () -> new StreamIterator(is);
	}

	@Override
	public long contentLength() {
	return -1;
	}
	}

	public static final class PublisherAdapter implements BodyPublisher {

	private final Publisher<? extends ByteBuffer> publisher;
	private final long contentLength;

	public PublisherAdapter(Publisher<? extends ByteBuffer> publisher,
	long contentLength) {
	this.publisher = Objects.requireNonNull(publisher);
	this.contentLength = contentLength;
	}

	@Override
	public final long contentLength() {
	return contentLength;
	}

	@Override
	public final void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
	publisher.subscribe(subscriber);
	}
	}
	}