Keep streams in JsonBinding open

src/main/java/org/eclipse/yasson/internal/JsonBinding.java

@@ -12,13 +12,15 @@
 
 package org.eclipse.yasson.internal;
 
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.io.Reader;
 import java.io.StringReader;
 import java.io.StringWriter;
 import java.io.Writer;
 import java.lang.reflect.Type;
+import java.nio.CharBuffer;
 import java.nio.charset.Charset;
 import java.util.Map;
 import java.util.Set;
@@ -73,15 +75,15 @@ public <T> T fromJson(String str, Type type) throws JsonbException {
 
     @Override
     public <T> T fromJson(Reader reader, Class<T> type) throws JsonbException {
-        try (JsonParser parser = jsonbContext.getJsonProvider().createParser(reader)) {
+        try (JsonParser parser = jsonbContext.getJsonProvider().createParser(new CloseSuppressingReader(reader))) {
             DeserializationContextImpl unmarshaller = new DeserializationContextImpl(jsonbContext);
             return deserialize(type, parser, unmarshaller);
         }
     }
 
     @Override
     public <T> T fromJson(Reader reader, Type type) throws JsonbException {
-        try (JsonParser parser = jsonbContext.getJsonProvider().createParser(reader)) {
+        try (JsonParser parser = jsonbContext.getJsonProvider().createParser(new CloseSuppressingReader(reader))) {
             DeserializationContextImpl unmarshaller = new DeserializationContextImpl(jsonbContext);
             return deserialize(type, parser, unmarshaller);
         }
@@ -119,7 +121,7 @@ public <T> T fromJsonStructure(JsonStructure jsonStructure, Type runtimeType) th
 
     private JsonParser inputStreamParser(InputStream stream) {
         return jsonbContext.getJsonParserFactory()
-                .createParser(stream,
+                .createParser(new CloseSuppressingInputStream(stream),
                               Charset.forName((String) jsonbContext.getConfig()
                                       .getProperty(JsonbConfig.ENCODING).orElse("UTF-8")));
     }
@@ -145,15 +147,15 @@ public String toJson(Object object, Type type) throws JsonbException {
     @Override
     public void toJson(Object object, Writer writer) throws JsonbException {
         final SerializationContextImpl marshaller = new SerializationContextImpl(jsonbContext);
-        try (JsonGenerator generator = writerGenerator(writer)) {
+        try (JsonGenerator generator = writerGenerator(new CloseSuppressingWriter(writer))) {
             marshaller.marshallWithoutClose(object, generator);
         }
     }
 
     @Override
     public void toJson(Object object, Type type, Writer writer) throws JsonbException {
         final SerializationContextImpl marshaller = new SerializationContextImpl(jsonbContext, type);
-        try (JsonGenerator generator = writerGenerator(writer)) {
+        try (JsonGenerator generator = writerGenerator(new CloseSuppressingWriter(writer))) {
             marshaller.marshallWithoutClose(object, generator);
         }
     }
@@ -226,12 +228,254 @@ private JsonGenerator streamGenerator(OutputStream stream) {
         Map<String, ?> factoryProperties = jsonbContext.createJsonpProperties(jsonbContext.getConfig());
         final String encoding = (String) jsonbContext.getConfig().getProperty(JsonbConfig.ENCODING).orElse("UTF-8");
         return jsonbContext.getJsonProvider().createGeneratorFactory(factoryProperties)
-                .createGenerator(stream, Charset.forName(encoding));
+                .createGenerator(new CloseSuppressingOutputStream(stream), Charset.forName(encoding));
     }
 
     @Override
     public void close() throws Exception {
         jsonbContext.getComponentInstanceCreator().close();
     }
 
+    /**
+     * {@link OutputStream} that suppresses {@link OutputStream#close()}.
+     */
+    static final class CloseSuppressingOutputStream extends OutputStream {
+
+        private final OutputStream delegate;
+
+        CloseSuppressingOutputStream(OutputStream delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void close() {
+            // suppress
+        }
+
+        @Override
+        public void write(int b) throws IOException {
+            delegate.write(b);
+        }
+
+        @Override
+        public void write(byte[] b) throws IOException {
+            delegate.write(b);
+        }
+
+        @Override
+        public void write(byte[] b, int off, int len) throws IOException {
+            delegate.write(b, off, len);
+        }
+
+    }
+
+    /**
+     * {@link InputStream} that suppresses {@link InputStream#close()}.
+     */
+    static final class CloseSuppressingInputStream extends InputStream {
+
+        private final InputStream delegate;
+
+        CloseSuppressingInputStream(InputStream delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void close() {
+            // suppress
+        }
+
+        @Override
+        public int read() throws IOException {
+            return delegate.read();
+        }
+
+        @Override
+        public int read(byte[] b) throws IOException {
+            return delegate.read(b);
+        }
+
+        @Override
+        public int read(byte[] b, int off, int len) throws IOException {
+            return delegate.read(b, off, len);
+        }
+
+        @Override
+        public byte[] readAllBytes() throws IOException {
+            return delegate.readAllBytes();
+        }
+
+        @Override
+        public byte[] readNBytes(int len) throws IOException {
+            return delegate.readNBytes(len);
+        }
+
+        @Override
+        public int readNBytes(byte[] b, int off, int len) throws IOException {
+            return delegate.readNBytes(b, off, len);
+        }
+
+        @Override
+        public long skip(long n) throws IOException {
+            return delegate.skip(n);
+        }
+
+        @Override
+        public int available() throws IOException {
+            return delegate.available();
+        }
+
+        @Override
+        public void mark(int readlimit) {
+            delegate.mark(readlimit);
+        }
+
+        @Override
+        public void reset() throws IOException {
+            delegate.reset();
+        }
+
+        @Override
+        public boolean markSupported() {
+            return delegate.markSupported();
+        }
+
+        @Override
+        public long transferTo(OutputStream out) throws IOException {
+            return delegate.transferTo(out);
+        }
+
+    }
+
+    /**
+     * {@link Reader} that suppresses {@link Reader#close()}.
+     */
+    static final class CloseSuppressingReader extends Reader {
+
+        private final Reader delegate;
+
+        CloseSuppressingReader(Reader delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public int read(CharBuffer target) throws IOException {
+            return delegate.read(target);
+        }
+
+        @Override
+        public int read() throws IOException {
+            return delegate.read();
+        }
+
+        @Override
+        public int read(char[] cbuf) throws IOException {
+            return delegate.read(cbuf);
+        }
+
+        @Override
+        public int read(char[] cbuf, int off, int len) throws IOException {
+            return delegate.read(cbuf, off, len);
+        }
+
+        @Override
+        public long skip(long n) throws IOException {
+            return delegate.skip(n);
+        }
+
+        @Override
+        public boolean ready() throws IOException {
+            return delegate.ready();
+        }
+
+        @Override
+        public boolean markSupported() {
+            return delegate.markSupported();
+        }
+
+        @Override
+        public void mark(int readAheadLimit) throws IOException {
+            delegate.mark(readAheadLimit);
+        }
+
+        @Override
+        public void reset() throws IOException {
+            delegate.reset();
+        }
+
+        @Override
+        public void close() {
+            // suppress
+        }
+
+        @Override
+        public long transferTo(Writer out) throws IOException {
+            return delegate.transferTo(out);
+        }
+
+    }
+
+    /**
+     * {@link Writer} that suppresses {@link Writer#close()}.
+     */
+    static final class CloseSuppressingWriter extends Writer {
+
+        private final Writer delegate;
+
+        CloseSuppressingWriter(Writer delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public void write(int c) throws IOException {
+            delegate.write(c);
+        }
+
+        @Override
+        public void write(char[] cbuf) throws IOException {
+            delegate.write(cbuf);
+        }
+
+        @Override
+        public void write(char[] cbuf, int off, int len) throws IOException {
+            delegate.write(cbuf, off, len);
+        }
+
+        @Override
+        public void write(String str) throws IOException {
+            delegate.write(str);
+        }
+
+        @Override
+        public void write(String str, int off, int len) throws IOException {
+            delegate.write(str, off, len);
+        }
+
+        @Override
+        public Writer append(CharSequence csq) throws IOException {
+            return delegate.append(csq);
+        }
+
+        @Override
+        public Writer append(CharSequence csq, int start, int end) throws IOException {
+            return delegate.append(csq, start, end);
+        }
+
+        @Override
+        public Writer append(char c) throws IOException {
+            return delegate.append(c);
+        }
+
+        @Override
+        public void flush() throws IOException {
+            delegate.flush();
+        }
+
+        @Override
+        public void close() {
+            // suppress
+        }
+
+    }
+
 }