PipedInputOutputStreamHandler.java

/*
 * GovWay - A customizable API Gateway 
 * https://govway.org
 * 
 * Copyright (c) 2005-2024 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.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.openspcoop2.utils.io.notifier.StreamingHandler;

/**
 * PipedInputOutputStreamHandler
 *
 * @author Poli Andrea (apoli@link.it)
 * @author $Author$
 * @version $Rev$, $Date$
 */
public class PipedInputOutputStreamHandler implements StreamingHandler {

	// Thread che si occupa di consumare i bytes disponibili forniti all'handler
	// Implementa anche la gestione dello stream tra producer e consumer
	private AbstractStreamingHandler streamingHandler;
	
	// Submit ritornato dell'esecuzione del thread
	private Future<ResultStreamingHandler> submitThreadExecution;
	
	// Risultato dell'esecuzione del thread
	private boolean retrieveResult = false;
	private ResultStreamingHandler resultThreadExecution;
	private String errorMessageThreadExecution;
	private Throwable exceptionThreadExecution;
	
	// Log
	private Logger log;
	
	// Esecutore del Thread
	private ExecutorService executor;
	
	// Informazione se lo stream is closed
	private boolean closed;
	
	// ID classe streamingHandler
	private String idStreamingHandler;
	
	public PipedInputOutputStreamHandler(String id, AbstractStreamingHandler streamingHandler, Logger log) throws Exception {
		
		this.log = log;
		
		//inizializzo la pipe di stream attraverso la quale eseguire la validazione
		//this.out = new PipedOutputStream();
		this.streamingHandler = streamingHandler;
		this.idStreamingHandler = id;
		
		// Creo esecutore del thread
		this.executor = Executors.newSingleThreadExecutor();
		
		// Avvio il thread
		this.submitThreadExecution = this.executor.submit(this.streamingHandler);
	}
	
	
	
	// ** Metodi interfaccia StreamingHandler **
	
	@Override
	public String getID(){
		return this.idStreamingHandler;
	}
	
	@Override
	public void feed(byte b) throws IOException {
		byte[]buffer = new byte[1];
		this.feed(buffer);
	}

	@Override
	public void feed(byte[] b) throws IOException {
		
		//System.out.println("@@PIPE@@ feed ["+b.length+"] bytes ...");
		
		try{
			if(!this.closed) {
				// in.isPrematureEnd() ci dice se il processo attuato dallo streaming handler e' gia' terminato
				// b == -1 significa che lo stream è finito
				if(this.streamingHandler.isPrematureEnd()){
					//System.out.println("@@PIPE@@ feed ["+b.length+"] bytes: END");
					this.end();
				}else {
					//System.out.println("@@PIPE@@ feed ["+b.length+"] bytes: WRITE");
					this.streamingHandler.write(b);
				}
			}
		}catch(Throwable e){
			this.log.error("["+this.idStreamingHandler+"] feed error",e);
			throw new IOException(e.getMessage());
		}
	}
	
	@Override
	public void end() throws IOException {
		
		//System.out.println("@@PIPE@@ END");
		closeResources();
		
	}
		
	@Override
	public void closeResources() throws IOException {
		
		//System.out.println("@@PIPE@@ CLOSE RESOURCES ...");
		IOException ioException = null;
		
		if(!this.closed) {
		
			try {
			
				// Chiudo input stream
				this.streamingHandler.close();
				
				// Fermo esecutore del thread
				this.executor.shutdown();				
				try {
	
					 if (!this.executor.awaitTermination(20, TimeUnit.SECONDS)) {
						 this.executor.shutdownNow();
					 }
				 } catch (InterruptedException pCaught) {	
					 this.executor.shutdownNow();
					 Thread.currentThread().interrupt();	
				 }
								
			}catch(Throwable e){
				if(e!=null && e instanceof InterruptedException) {
					Thread.currentThread().interrupt();	
				}
				this.log.error("["+this.idStreamingHandler+"] end error",e);
				ioException = new IOException(e.getMessage());
			}
		}
		
		//System.out.println("@@PIPE@@ RETRIVE RESULT ...");
		retrieveResult();
		//System.out.println("@@PIPE@@ RETRIVED RESULT");
		
		//System.out.println("@@PIPE@@ Release resource...");
		this.releaseResource(); 
		//System.out.println("@@PIPE@@ Release resource ok");
		
		if(ioException!=null){
			throw ioException;
		}
		
	}

	private void retrieveResult() {
		
		if(this.retrieveResult == false){
			try {
								
				// Recupero risultato
				if(this.submitThreadExecution!=null){
					//System.out.println("@@PIPE@@ RETRIVE RESULT submitThreadExecution.get()...");
					this.resultThreadExecution = this.submitThreadExecution.get();
					//System.out.println("@@PIPE@@ RETRIVE RESULT submitThreadExecution.get() ok");
				}
				
				// Recupero eventuali errori non lanciati nella exception del metodo call() (Utile negli scenari in cui l'errore viene capito in streaming all'interno del thread)
				this.errorMessageThreadExecution = this.streamingHandler.getError();
				this.exceptionThreadExecution = this.streamingHandler.getException();
								
			}catch(Throwable e){
				
				if(e !=null && e instanceof InterruptedException) {
					Thread.currentThread().interrupt();
				}
				
				// Errori generati da this.submitThreadExecution.get()
				// Cosi' facendo recupero eventuali errori lanciati nella exception del metodo call()
				this.errorMessageThreadExecution = e.getMessage();
				this.exceptionThreadExecution = e;
				
				this.log.error("["+this.idStreamingHandler+"] end error",e);
				// Non devo rilanciare gli errori, la logica e' rimandata a chi usa gli handler.
				// Devo solo salvare gli errori
				// throw new IOException(e.getMessage());
			}
			finally{
				// Risultati letti
				this.retrieveResult = true;
			}
		}
	}
	
	private void releaseResource() throws IOException{
		try {
			if(!this.closed) {
				//this.out = null;
				this.streamingHandler.close();
				this.streamingHandler = null;
				this.submitThreadExecution = null;
				this.closed = true;
			}
		}catch(Exception e){
			this.log.error("["+this.idStreamingHandler+"] closeResources error",e);
			throw new IOException(e.getMessage());
		}
	}
	
	
	
	
	
	// ** Metodi Recuperare informazioni elaborate dallo streaming handler **
	
	private void finalizeResult() throws IOException{
		// La close puo' non essere ancora stata effettuato:
		// - puo' succedere che non sia stato chiamato l'end perche' il messaggio e' piccolo e viene gestito in una unica feed.
		// - inoltre il metodo closeResources puo' essere invocato alla fine di tutto (es. dopo aver ritornato la risposta)
		if(!this.closed) {
			closeResources();
		}
	}
	
	public ResultStreamingHandler getResult() throws IOException {
		this.finalizeResult();
		return this.resultThreadExecution;
	}

	public String getError() throws IOException {
		this.finalizeResult();
		return this.errorMessageThreadExecution;
	}
	
	public Throwable getException() throws IOException {
		this.finalizeResult();
		return this.exceptionThreadExecution;
	}

	



}