PipedBytesStream.java

/*
 * GovWay - A customizable API Gateway 
 * https://govway.org
 * 
 * Copyright (c) 2005-2025 Link.it srl (https://link.it). 
 * 
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 3, as published by
 * the Free Software Foundation.
 *
 * This program 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 for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 */
package org.openspcoop2.utils.io.notifier.unblocked;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import org.openspcoop2.utils.SemaphoreLock;
import org.openspcoop2.utils.Utilities;
import org.slf4j.Logger;

/**
 * PipedBytesStream
 *
 * @author $Author$
 * @version $Rev$, $Date$
 */
public class PipedBytesStream extends IPipedUnblockedStream {
    private static final int MAX_QUEUE = 2;

    protected Logger log = null;
    private long sizeBuffer;
    private int timeoutMs;
    
    @Override
    public void init(Logger log, long sizeBuffer, int timeoutMs, String source) {
        this.log = log;
        // In memoria esistono 2 buffer, 
        // - [bytesReceived] uno che contiene i bytes gia' consolidati pronti a essere consumati
        // - [bytesReading] buffer utilizzato per scrivere i dati
        // Quindi puo' succedere che entrambi i buffer siano "pieni". La dimensione massima richiesta in memoria viene quindi divisa per 2.
        // NOTA: La dimensione di ogni buffer potra' essere this.sizeBuffer + eventualmente ;a dimensione del byte[] fornita con l'ultima write che ha superato il controllo di waitSpaceAvailable 
        if(sizeBuffer<=0) {
            sizeBuffer = Utilities.DIMENSIONE_BUFFER;
        }
        this.sizeBuffer = sizeBuffer / 2; 
        this.timeoutMs = timeoutMs;
        this.source = source;
        initReadingBuffer();
    }
    @Override
    public void setTimeout(int timeoutMs) {
        this.timeoutMs = timeoutMs;
    }
    
    //private final Integer semaphore = 1;
    private final org.openspcoop2.utils.Semaphore lockPIPE = new org.openspcoop2.utils.Semaphore("PipedBytesStream");
    private byte [] bytesReading = null;
    //private volatile int indexNextByteReceivedForWrite = 0;
    private int indexNextByteReceivedForWrite = 0;
    private List<byte[]> chunkList = new ArrayList<byte[]>();
    private byte [] bytesReceived = null;
    //private volatile int indexNextByteReceivedForRead = -1;
    private int indexNextByteReceivedForRead = -1;
    
    private boolean stop = false;

    private boolean useThreadSleep = false;
    private static final int ITERAZIONI_WAIT = 128;
    private CompletableFuture<Boolean> asyncReadTask = null;
    private CompletableFuture<Boolean> asyncWriteTask = null;

    private void initReadingBuffer() {
        this.bytesReading = new byte[ (int)this.sizeBuffer ];
        this.indexNextByteReceivedForWrite = 0;
    }

    private String source = null;
    public String getPrefixSource() {
        return this.source!=null ? "["+this.source+"] " : "";
    }

    private boolean readBytesPending() {
        return (this.indexNextByteReceivedForWrite > 0 || this.chunkList.size() > 0);
    }

    // INPUT STREAM
    

    private void readWaitBytes() throws IOException{
        try {
            if(this.useThreadSleep) {
                int i = 0;
                while(this.stop==false && !readBytesPending() && i<ITERAZIONI_WAIT){
                    Utilities.sleep((i+1));
                    i = i + i;
                }
                if(i>=ITERAZIONI_WAIT){
                    throw new IOException(getPrefixSource()+"Timeout, no bytes available for read");
                }
            } else {
                boolean wait = false;
                SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("readWaitBytes");
                try {
                    if(this.stop==false && !readBytesPending()) {
                        this.asyncReadTask = new CompletableFuture<Boolean>();
                        wait = true;
                    }
                }finally {
                    this.lockPIPE.release(lock, "readWaitBytes");
                }
                
                if(wait) {
                    try {
    //                  System.out.println("["+this.source+"] ASPETTO READ...");
                        if(this.timeoutMs>0) {
                            this.asyncReadTask.get(this.timeoutMs,TimeUnit.MILLISECONDS );
                        }
                        else {
                            this.asyncReadTask.get();
                        }
    //                  System.out.println("["+this.source+"] READ OK");
                    }catch (InterruptedException timeout) {
                        Thread.currentThread().interrupt();
                        throw new IOException(getPrefixSource()+"Timeout, no bytes available for read: "+timeout.getMessage(),timeout);
                    }catch(Exception timeout) {
                        throw new IOException(getPrefixSource()+"Timeout, no bytes available for read: "+timeout.getMessage(),timeout);
                    }
                }
            }
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            if(t !=null && t instanceof InterruptedException) {
                Thread.currentThread().interrupt();
            }
            throw new IOException(t.getMessage(),t);
        }
    }
    
    @Override
    public int read(byte[] b) throws IOException {
        
        return this.read(b, 0, b.length);
        
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException {
        
        try {
            
    //      System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] .....");
            //this.log.debug("########### READ b["+b.length+"] off["+off+"] len["+len+"] .....");
            
            if(this.bytesReceived==null){
                if (this.stop) {        
                    if(!readBytesPending()){
    //                  System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] STOP BOUT NULL return -1");
                        if(this.asyncWriteTask!=null) {
    //                      System.out.println("["+this.source+"] READ for WRITE COMPLETE 1");
                            this.asyncWriteTask.complete(true);
                        }
                        return -1;
                    }
                } else {
                    if(!readBytesPending()){
    //                  System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] WAIT BYTES ...");
                        readWaitBytes();
                        if (!readBytesPending()) {
                            // Viene reso null dal metodo close() che puo' essere chiamato mentre la read e' in corso
    //                      System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] WAIT BYTES FOUND BOUT NULL ON EXIT");
                            if(this.asyncWriteTask!=null) {
    //                          System.out.println("["+this.source+"] READ for WRITE COMPLETE 3");
                                this.asyncWriteTask.complete(true);
                            }
                            return -1;
                        }
                    }
                }
            }
            
            //this.log.debug("########### READ b["+b.length+"] off["+off+"] len["+len+"] BYTES AVAILABLE ...");
            
            if(this.bytesReceived==null){
                if(this.stop){
                    if(!readBytesPending()){
    //                  System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] BYTES AVAILABLE RETURN -1");
                        if(this.asyncWriteTask!=null) {
    //                      System.out.println("["+this.source+"] READ for WRITE COMPLETE 4");
                            this.asyncWriteTask.complete(true);
                        }
                        return -1;
                    }
                }
            }
            
                    
            
            
            if(this.bytesReceived==null){
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] BYTES AVAILABLE FROM PRECEDENT BUFFERING IS NULL ...");
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] SYNC ...");
                //synchronized (this.semaphore) {
                SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("read");
                try {
    //              System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] CHUNKS ... " + this.chunkList.size() );
    //              System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] Next-Byte for write ... " + this.indexNextByteReceivedForWrite );
                    if ( this.chunkList.size() > 0 ) {
                        this.bytesReceived = this.chunkList.remove(0);
                    } else
                    if ( this.indexNextByteReceivedForWrite > 0 ) {
                        this.bytesReceived = new byte[ this.indexNextByteReceivedForWrite ];
                        System.arraycopy( this.bytesReading, 0, this.bytesReceived, 0, this.indexNextByteReceivedForWrite );
                        this.indexNextByteReceivedForWrite = 0;
                    }
                    this.indexNextByteReceivedForRead = 0;
                    if(this.asyncWriteTask!=null) {
    //                  System.out.println("["+this.source+"] READ for WRITE COMPLETE IN SEMAPHORE");
                        this.asyncWriteTask.complete(true);
                    }
                } finally {
                    this.lockPIPE.release(lock, "read");
                }
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] SYNC OK");
            }
            
            if(this.bytesReceived==null){
                return -1;
            }
            
            int bytesAvailableForRead = this.bytesReceived.length - this.indexNextByteReceivedForRead;
            if ( bytesAvailableForRead == len ) {
                System.arraycopy( this.bytesReceived, this.indexNextByteReceivedForRead, b, off, bytesAvailableForRead );
                this.bytesReceived = null;
                this.indexNextByteReceivedForRead = -1;
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] NEXT INDEX["+this.indexNextByteReceivedForRead+"] RETURN LETTI("+len+") EXIT A");
                return len;
            } else
            if ( bytesAvailableForRead > len ) {
                System.arraycopy( this.bytesReceived, this.indexNextByteReceivedForRead, b, off, len );
                this.indexNextByteReceivedForRead += len;
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] NEXT INDEX["+this.indexNextByteReceivedForRead+"] RETURN LETTI("+len+") EXIT B");
                return len;
            } else {
                System.arraycopy( this.bytesReceived, this.indexNextByteReceivedForRead, b, off, bytesAvailableForRead );
                this.bytesReceived = null;
                this.indexNextByteReceivedForRead = -1;
    //          System.out.println("########### READ b["+b.length+"] off["+off+"] len["+len+"] NEXT INDEX["+this.indexNextByteReceivedForRead+"] RETURN LETTI("+bytesAvailableForRead+") EXIT C");
                return bytesAvailableForRead;
            }
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            throw new IOException(t.getMessage(),t);
        }
    }

    @Override
    public int read() throws IOException {
        byte[]b = new byte[1];
        int len = this.read(b);
        if ( len == 1 )
            return b[0] & 0xFF;
        if ( len == -1 )
            return -1;
        throw new IOException( "Cannot read single byte" );
    }

    
    @Override
    public void close() throws IOException {
        try {
            if ( this.stop == false ) {
    
                SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("close");
                try {
                    if ( this.indexNextByteReceivedForWrite > 0 ) {
                        byte[] lastChunk = new byte[ this.indexNextByteReceivedForWrite ];
                        System.arraycopy( this.bytesReading, 0, lastChunk, 0, this.indexNextByteReceivedForWrite );
                        this.chunkList.add( lastChunk );
                        this.indexNextByteReceivedForWrite = 0;
                    }
                    this.bytesReading = null;
                    this.stop = true;
                }finally{
                    this.lockPIPE.release(lock, "close");
                }
            }
            if(this.asyncWriteTask!=null) {
    //          System.out.println("["+this.source+"] CLOSE for WRITE COMPLETE");
                this.asyncWriteTask.complete(true);
            }
            if(this.asyncReadTask!=null) {
    //          System.out.println("["+this.source+"] CLOSE for READ COMPLETE");
                this.asyncReadTask.complete(true);
            }
        }
        catch(Throwable t) {
            throw new IOException(t.getMessage(),t);
        }
    }

    
    // ALIMENTAZIONE
    
    private void writeWaitEmptyBuffer() throws IOException{
        try {
            if ( this.useThreadSleep ) {
                if ( readBytesPending() && this.chunkList.size() >= MAX_QUEUE ) {
                    int i = 0;
                    while ( this.stop == false && i < ITERAZIONI_WAIT ) {
                        Utilities.sleep((i+1));
                        i = i + i;
                    }
                    if ( i >= ITERAZIONI_WAIT ) {
                        throw new IOException(getPrefixSource()+"Timeout, no buffer available for write");
                    }
                }
            } else {
                boolean wait = false;
                SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("writeWaitEmptyBuffer");
                try {
                    if ( readBytesPending() && this.chunkList.size() >= MAX_QUEUE ) {
                        if ( this.stop == false ) {
                            this.asyncWriteTask = new CompletableFuture<Boolean>();
                            wait = true;
                        }
                    }
                }finally{
                    this.lockPIPE.release(lock, "writeWaitEmptyBuffer");
                }
                if(wait) {
                    try {
    //                  System.out.println("["+this.source+"] ASPETTO WRITE ...");
                        if(this.timeoutMs>0) {
                            this.asyncWriteTask.get(this.timeoutMs,TimeUnit.MILLISECONDS );
                        }
                        else {
                            this.asyncWriteTask.get();
                        }
    //                  System.out.println("["+this.source+"] WRITE OK");
                    }catch(InterruptedException timeout) {
                        Thread.currentThread().interrupt();
                        throw new IOException(getPrefixSource()+"Timeout, no bytes available for read: "+timeout.getMessage(),timeout);
                    }
                    catch(Exception timeout) {
                        throw new IOException(getPrefixSource()+"Timeout, no bytes available for read: "+timeout.getMessage(),timeout);
                    }
                }
            }
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            if(t !=null && t instanceof InterruptedException) {
                Thread.currentThread().interrupt();
            }
            throw new IOException(t.getMessage(),t);
        }
    }
    
    @Override
    public void write(byte b) throws IOException{
        
        try {
        
            //this.log.debug("########### WRITE byte .....");
            
            if ( this.bytesReading == null ) {
                throw new IOException(getPrefixSource()+"Stream already closed");
            }
            
            this.writeWaitEmptyBuffer();
                    
            //this.log.debug("########### WRITE byte SYNC ...");
            //synchronized (this.semaphore) {
            SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("write(b)");
            try {
                this.bytesReading[ this.indexNextByteReceivedForWrite++ ] = b;
                if ( this.indexNextByteReceivedForWrite == this.bytesReading.length ) {
                    this.chunkList.add( this.bytesReading );
                    initReadingBuffer();
                }
                if(this.asyncReadTask!=null) {
    //              System.out.println("["+this.source+"] WRITE for READ COMPLETE 1");
                    this.asyncReadTask.complete(true);
                }
                //this.log.debug("########### WRITE byte SYNC OK");
    //          System.out.println("########### WRITE byte SYNC OK ADD[1] SIZE_ATTUALE["+ this.indexNextByteReceivedForWrite + " - " + this.chunkList.size()+"]");
            }finally{
                this.lockPIPE.release(lock, "write(b)");
            }
            
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            throw new IOException(t.getMessage(),t);
        }

    }
    
    @Override
    public void write(byte [] b) throws IOException{
        
        try {
        
            //this.log.debug("########### WRITE byte ["+b.length+"] .....");
            
            if ( this.bytesReading == null ) {
                throw new IOException(getPrefixSource()+"Stream already closed");
            }
            
            this.writeWaitEmptyBuffer();
                    
            //this.log.debug("########### WRITE byte ["+b.length+"] SYNC ...");
            //synchronized (this.semaphore) {
            SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("write(b[])");
            try {
                int offset = 0;
                int bytesNum = b.length;
                while ( bytesNum > 0 ) {
                    int freeSpace = this.bytesReading.length - this.indexNextByteReceivedForWrite;
                    int chunkLen = ( bytesNum <= freeSpace ? bytesNum : freeSpace );
                    System.arraycopy( b, offset, this.bytesReading, this.indexNextByteReceivedForWrite, chunkLen );
                    this.indexNextByteReceivedForWrite += chunkLen;
                    if ( this.indexNextByteReceivedForWrite == this.bytesReading.length ) {
                        this.chunkList.add( this.bytesReading );
                        initReadingBuffer();
                    }
                    bytesNum -= chunkLen;
                    offset += chunkLen;
                }
                if(this.asyncReadTask!=null) {
    //              System.out.println("["+this.source+"] WRITE for READ COMPLETE 2");
                    this.asyncReadTask.complete(true);
                }
                //this.log.debug("########### WRITE byte ["+b.length+"] SYNC OK");
    //          System.out.println("########### WRITE byte SYNC OK ADD["+b.length+"] SIZE_ATTUALE["+ this.indexNextByteReceivedForWrite + " - " + this.chunkList.size()+"]");
            }finally {
                this.lockPIPE.release(lock, "write(b[])");
            }
            
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            throw new IOException(t.getMessage(),t);
        }

    }
    
    @Override
    public void write(byte[] b, int off, int len) throws IOException{
        
        try {
        
            //this.log.debug("########### WRITE byte ["+b.length+"] off:"+off+" len:"+len+" .....");
            
            if ( this.bytesReading == null ) {
                throw new IOException(getPrefixSource()+"Stream already closed");
            }
            
            this.writeWaitEmptyBuffer();
                    
            //this.log.debug("########### WRITE byte ["+b.length+"] off:"+off+" len:"+len+" SYNC ...");
            //synchronized (this.semaphore) {
            SemaphoreLock lock = this.lockPIPE.acquireThrowRuntime("write(b[],off,len)");
            try {
                int offset = off;
                int bytesNum = len;
                while ( bytesNum > 0 ) {
                    int freeSpace = this.bytesReading.length - this.indexNextByteReceivedForWrite;
                    int chunkLen = ( bytesNum <= freeSpace ? bytesNum : freeSpace );
                    System.arraycopy( b, offset, this.bytesReading, this.indexNextByteReceivedForWrite, chunkLen );
                    this.indexNextByteReceivedForWrite += chunkLen;
                    if ( this.indexNextByteReceivedForWrite == this.bytesReading.length ) {
                        this.chunkList.add( this.bytesReading );
                        initReadingBuffer();
                    }
                    bytesNum -= chunkLen;
                    offset += chunkLen;
                }
                if(this.asyncReadTask!=null) {
    //              System.out.println("["+this.source+"] WRITE for READ COMPLETE 3");
                    this.asyncReadTask.complete(true);
                }
                //this.log.debug("########### WRITE byte ["+b.length+"] off:"+off+" len:"+len+" SYNC OK");
    //          System.out.println("########### WRITE byte SYNC OK ADD["+b.length+"] SIZE_ATTUALE["+ this.indexNextByteReceivedForWrite + " - " + this.chunkList.size()+"]");
            }finally {
                this.lockPIPE.release(lock, "write(b[],off,len)");
            }
            
        }
        catch(IOException io) {
            throw io;
        }
        catch(Throwable t) {
            throw new IOException(t.getMessage(),t);
        }

    }

}