NotifierInputStream.java

  1. /*
  2.  * GovWay - A customizable API Gateway 
  3.  * https://govway.org
  4.  * 
  5.  * Copyright (c) 2005-2025 Link.it srl (https://link.it). 
  6.  * 
  7.  * This program is free software: you can redistribute it and/or modify
  8.  * it under the terms of the GNU General Public License version 3, as published by
  9.  * the Free Software Foundation.
  10.  *
  11.  * This program is distributed in the hope that it will be useful,
  12.  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13.  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14.  * GNU General Public License for more details.
  15.  *
  16.  * You should have received a copy of the GNU General Public License
  17.  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18.  *
  19.  */
  20. package org.openspcoop2.utils.io.notifier;

  22. import java.io.ByteArrayOutputStream;
  23. import java.io.IOException;
  24. import java.io.InputStream;
  25. import java.io.OutputStream;
  26. import java.util.ArrayList;
  27. import java.util.HashMap;
  28. import java.util.List;
  29. import java.util.Map;

  31. import org.slf4j.Logger;

  33. import com.sun.xml.messaging.saaj.packaging.mime.internet.ContentType;
  34. import com.sun.xml.messaging.saaj.packaging.mime.internet.ParseException;

  36. /**
  37.  * NotifierInputStream
  38.  *
  39.  * @author Poli Andrea (apoli@link.it)
  40.  * @author $Author$
  41.  * @version $Rev$, $Date$
  42.  */
  43. public class NotifierInputStream extends InputStream {

  45.     
  46.     /* ****** VARIABLE ******* */
  47.     
  48.     /**
  49.      * Original Input Stream
  50.      */
  51.     private InputStream is; 
  52.     /**
  53.      * Original Input Stream Consumed
  54.      */
  55.     private boolean isOriginalInputStreamConsumed;
  56.     
  57.     /**
  58.      * Current position in this input stream
  59.      */
  60.     private int currentReadPositionFromStream = 0;
  61.     
  62.     /**
  63.      * Content type of the stream
  64.      */
  65.     private ContentType contentType = null;
  66.     
  67.     /**
  68.      * Indication if the buffer is enabled
  69.      */
  70.     private boolean bufferEnabled = false;
  71.     public boolean isBufferEnabled() {
  72.         return this.bufferEnabled;
  73.     }

  75.     /**
  76.      * Buffer
  77.      */
  78.     private ByteArrayOutputStream buffer = null;
  79.     /**
  80.      * Bytes present in the stream after the completion of reading
  81.      */
  82.     private byte[] contentCompleteReadedFromStream = null;  
  83.     
  84.     /**
  85.      * StreamingHandler
  86.      */
  87.     private Map<String,StreamingHandler> streamingHandlers = new HashMap<String, StreamingHandler>();
  88.     private List<String> streamingHandlersIds = new ArrayList<>(); // Per preservare l'ordine di inserimento

  90.     
  91.     /**
  92.      * throwStreamingHandlerException
  93.      */
  94.     private boolean throwStreamingHandlerException = false;
  95.     
  96.     /**
  97.      * InputStream is closed
  98.      */
  99.     private boolean isClosed = false;
  100.     
  101.     
  102.     /**
  103.      * Logger
  104.      */
  105.     private Logger log = null;
  106.     
  107.     
  108.     /* ****** CONSTRUCTOR ******* */
  109.     
  110.     public NotifierInputStream(InputStream is,String contentType,NotifierInputStreamParams params) throws IOException, ParseException{
  111.         
  112.         // Set content type of the stream
  113.         //try{
  114.         if(contentType!=null){
  115.             this.contentType = new ContentType(contentType);
  116.         }
  117.         //}catch(Exception e){
  118.         //  throw new IOException(e.getMessage(),e);
  119.         //}
  120.         
  121.         // Set original input stream
  122.         if(is==null){
  123.             throw new ParseException("Original InputStream not defined in args");
  124.         }
  125.         this.is = is;
  126.         //System.out.println("@@@@@@@@ NotifierInputStream: "+is.getClass().getName());
  127.                 
  128.         // Initialize Streaming Handler List
  129.         if(params.sizeStreamingHandlers()>0){
  130.             for (String streamingHandlerId : params.getStreamingHandlerIds()) {
  131.                 try{
  132.                     //System.out.println("@@@@@@@@ INIT HANDLERS");
  133.                     this.streamingHandlers.put(streamingHandlerId, params.getStreamingHandler(streamingHandlerId));
  134.                     this.streamingHandlersIds.add(streamingHandlerId);
  135.                 }catch(Exception e){
  136.                     throw new IOException("Streaming Handler initialization failed (id:"+streamingHandlerId+")"); 
  137.                 }
  138.             }
  139.         }
  140.         
  141.         // throwStreamingHandlerException
  142.         this.throwStreamingHandlerException = params.isThrowStreamingHandlerException();
  143.         
  144.         // Log
  145.         this.log = params.getLog();
  146.         
  147.         // Buffering
  148.         if(params.isBufferEnabled()){
  149.             this.setONBuffering();
  150.         }
  151.         
  152.     }
  153.     
  154.     
  155.     
  156.     
  157.     
  158.     
  159.     /* ****** GET ******* */

  161.     public ContentType getContentType() {
  162.         return this.contentType;
  163.     }


  166.     
  167.     
  168.     
  169.     
  170.     /* ******* INPUT STREAM INTERFACE METHODS ******* */
  171.     
  172.     @Override
  173.     public int read(byte[] b) throws IOException {
  174.         return this.read(b, 0, b.length);
  175.     }
  176.     
  177.     
  178.     @Override
  179.     public int read(byte[] b, int off, int len) throws IOException { 
  180.         return this.read_engine(b, off, len, true);
  181.     }
  182.     
  183. //  java.io.FileOutputStream fout = null;
  184. //  java.io.File f = null;
  185. //  private void debug(byte[] b, int off, int letti){
  186. //      try{
  187. //          if(this.fout==null){
  188. //              this.f = java.io.File.createTempFile("NotifierInputStream", "tmp");
  189. //              this.fout = new java.io.FileOutputStream(this.f);
  190. //              this.fout.write(b, off, letti);
  191. //          }
  192. //      }catch(Exception e){
  193. //          System.out.println("@@@@@@@@ DEBUG ERROR:"+e.getMessage());
  194. //      }
  195. //  }
  196. //  private void printFile(){
  197. //      try{
  198. //          this.fout.flush();
  199. //          this.fout.close();
  200. //      }catch(Exception e){
  201. //          System.out.println("@@@@@@@@ FILE ERROR:"+e.getMessage());
  202. //      }
  203. //      System.out.println("@@@@@@@@ FILE["+this.f.getAbsolutePath()+"]");
  204. //  }
  205.     
  206.     private int read_engine(byte[] b, int off, int len, boolean incrementCurrentReadPositionFromStream) throws IOException {
  207.         
  208. //      if(!incrementCurrentReadPositionFromStream){
  209. //          System.out.println("@@@@@@@@ READ FROM SERIALIZE b["+b.length+"] offset["+off+"] length["+len+"]");
  210. //      }
  211. //      else{
  212. //          System.out.println("@@@@@@@@ READ b["+b.length+"] offset["+off+"] length["+len+"]");
  213. //      }
  214.         
  215.         int readBytes = 0;
  216.         int offset = off;
  217.         try{
  218.             
  219.             if(this.isOriginalInputStreamConsumed){
  220.             
  221.                 //System.out.println("@@@@@@@@ IS COMPLITED.... ");
  222.                 
  223.                 // The stream was completely consumed and the bytes are stored in the variable 'contentReadFromStream'
  224.                 if(this.contentCompleteReadedFromStream!=null){
  225.                     
  226. //                  System.out.println("@@@@@@@@ IS COMPLITED currentReadPositionFromStream["+this.currentReadPositionFromStream
  227. //                          +"] contentCompleteReadedFromStream["+this.contentCompleteReadedFromStream.length+"].... ");
  228.                     
  229.                     if(this.currentReadPositionFromStream>=this.contentCompleteReadedFromStream.length){
  230.                         // stream completed
  231. //                      System.out.println("@@@@@@@@ IS COMPLITED RETURN -1");
  232. //                      this.printFile();
  233.                         return -1; 
  234.                     }
  235.                     else{
  236.                         int position = this.currentReadPositionFromStream;
  237.                         while( (position<this.contentCompleteReadedFromStream.length) && (readBytes<len) ){
  238.                             
  239.                             if(offset>=b.length){
  240.                                 throw new IndexOutOfBoundsException("Offset: "+offset+" , byte[] length: "+b.length);
  241.                             }
  242.                             
  243.                             b[offset] = this.contentCompleteReadedFromStream[position];
  244.                             readBytes++;
  245.                             offset++;
  246.                         }
  247. //                      System.out.println("@@@@@@@@ IS COMPLITED RETURN "+readBytes+" BYTES");
  248. //                      this.debug(b, offset, readBytes);
  249.                         return readBytes;
  250.                     }
  251.                     
  252.                 } 
  253.                 
  254.                 // The stream has already been completely read. The error may be due to the use of the methods serialize without buffering enabled
  255.                 else{
  256.                     //throw new IOException("The stream has already been completely read. The error may be due to the use of the methods serialize without buffering enabled");
  257.                     // NOTE: the engine can call read method more times ...
  258.                     return -1;
  259.                 }
  260.                 
  261.             }
  262.             
  263.             // Reading in the progress of the stream
  264.             else{
  265.                                 
  266.                 // performRead: enable dispatching to streaming handlers
  267.                 readBytes = performReadBytes(b,off,len);
  268.                 
  269.                 // stream completed
  270.                 if(readBytes == -1){
  271.                     //System.out.println("@@@@@@@@ IS complite ");
  272.                     this.isOriginalInputStreamConsumed = true;
  273.                 }
  274.                 
  275.                 // if enabled buffering, the byte read is saved
  276.                 if(this.bufferEnabled){
  277.                     if(this.isOriginalInputStreamConsumed){
  278.                         //System.out.println("@@@@@@@@ IS finalize! ");
  279.                         this.finalizeBuffer();
  280.                     }else{
  281.                         //System.out.println("@@@@@@@@ write offset["+offset+"] readBytes["+readBytes+"] ");
  282.                         this.buffer.write(b, offset, readBytes);
  283.                     }
  284.                 }
  285.                 
  286. //              // Check if exists more bytes (questi codice non dovrebbe servire)
  287. //              int bytesMaxRead = len-off;
  288. //              if(readBytes<bytesMaxRead){
  289. //                  //System.out.println("@@@@@@@@ Check if exists more bytes ...");
  290. //                  int byteRead = read_engine(true);
  291. //                  //System.out.println("@@@@@@@@ Check if exists more bytes, return: "+byteRead);
  292. //                  if(byteRead!=-1){
  293. //                      //System.out.println("@@@@@@@@  Exists more byte, set return at position ["+readBytes+"]");
  294. //                      b[readBytes]=(byte)byteRead;
  295. //                      readBytes++;
  296. //                  }
  297. //              }
  298.                 
  299.                 //System.out.println("@@@@@@@@ Return dopo Perform "+readBytes);
  300. //              if(readBytes == -1){
  301. //                  this.printFile();
  302. //              }else{
  303. //                  this.debug(b, offset, readBytes);
  304. //              }           
  305.                 return readBytes;
  306.             }
  307.         }finally{
  308.             if(incrementCurrentReadPositionFromStream){
  309.                 this.currentReadPositionFromStream=this.currentReadPositionFromStream+readBytes;
  310.             }
  311.         }
  312.     }
  313.     
  314.     private int performReadBytes(byte[] b, int off, int len) throws IOException {
  315.         
  316.         //System.out.println("@@@@@@@@ performReadBytes ...");
  317.         
  318.         int readBytes = 0;
  319.         
  320.         if(this.is==null){
  321.             //System.out.println("@@@@@@@@ return -1 Stream is null ...");
  322.             readBytes = -1;
  323.         }
  324.         else{
  325.             readBytes = this.is.read(b,off,len);
  326.         }
  327.         //System.out.println("@@@@@@@@ ["+readBytes+"] bytes read ...");
  328.         
  329.         // enable dispatching to streaming handlers
  330.         ByteArrayOutputStream bout = null;
  331.         for(String streamingHandlerId : this.streamingHandlersIds) {
  332.             //System.out.println("@@@@@@@@ Streaming handler ["+this.streamingHandlersIds.size()+"]");
  333.             StreamingHandler streamingHandler = this.streamingHandlers.get(streamingHandlerId);
  334.             try{
  335.                 if(readBytes==-1){
  336.                     //System.out.println("@@@@@@@@ return -1 Dispatching end...");
  337.                     streamingHandler.end();
  338.                 }
  339.                 else{
  340.                     if(bout==null){
  341.                         bout = new ByteArrayOutputStream();
  342.                         bout.write(b, off, readBytes);
  343.                         bout.flush();
  344.                         bout.close();
  345.                     }
  346.                     //System.out.println("@@@@@@@@ return bytes "+bout.size()+" Dispatching ...");
  347.                     streamingHandler.feed(bout.toByteArray());
  348.                 }
  349.             }catch(Throwable e){
  350.                 if(this.log!=null){
  351.                     this.log.error("["+streamingHandlerId+"] error occurs: "+e.getMessage(),e);
  352.                 }
  353.                 if(this.throwStreamingHandlerException){
  354.                     throw new IOException("["+streamingHandlerId+"] "+e.getMessage());
  355.                 }
  356.             }
  357.         }
  358.                 
  359.         return readBytes;
  360.     }
  361.     


  364.     
  365.     @Override
  366.     public int read() throws IOException {
  367.         return this.read_engine(true);
  368.     }
  369.     private int read_engine(boolean incrementCurrentReadPositionFromStream) throws IOException {
  370.         try{
  371.             
  372. //          if(!incrementCurrentReadPositionFromStream){
  373. //              System.out.println("@@@@@@@@ READ FROM SERIALIZE");
  374. //          }
  375. //          else{
  376. //              System.out.println("@@@@@@@@ READ");
  377. //          }
  378.             
  379.             if(this.isOriginalInputStreamConsumed){
  380.             
  381.                 //System.out.println("@@@@@@@@ IS COMPLITED SINGLE BYTE.... ");
  382.                 
  383.                 // The stream was completely consumed and the bytes are stored in the variable 'contentReadFromStream'
  384.                 if(this.contentCompleteReadedFromStream!=null){
  385.                     
  386. //                  System.out.println("@@@@@@@@ IS COMPLITED SINGLE BYTE currentReadPositionFromStream["+this.currentReadPositionFromStream
  387. //                          +"] contentCompleteReadedFromStream["+this.contentCompleteReadedFromStream.length+"].... ");
  388.                     
  389.                     if(this.currentReadPositionFromStream>=this.contentCompleteReadedFromStream.length){
  390.                         // stream completed
  391. //                      System.out.println("@@@@@@@@ IS COMPLITED SINGLE BYTE RETURN -1");
  392. //                      this.printFile();
  393.                         return -1; 
  394.                     }
  395.                     else{
  396. //                      System.out.println("@@@@@@@@ IS COMPLITED RETURN SINGLE BYTE ("+this.contentCompleteReadedFromStream[this.currentReadPositionFromStream]+") ");
  397. //                      this.debug(new byte[]{this.contentCompleteReadedFromStream[this.currentReadPositionFromStream]}, 0, 1);
  398.                         return this.contentCompleteReadedFromStream[this.currentReadPositionFromStream];
  399.                     }
  400.                     
  401.                 } 
  402.                 
  403.                 // The stream has already been completely read. The error may be due to the use of the methods serialize without buffering enabled
  404.                 else{
  405.                     //throw new IOException("The stream has already been completely read. The error may be due to the use of the methods serialize without buffering enabled");
  406.                     // NOTE: the engine can call read method more times ...
  407.                     return -1;
  408.                 }
  409.                 
  410.             }
  411.             
  412.             // Reading in the progress of the stream
  413.             else{
  414.                                 
  415.                 // performRead: enable dispatching to streaming handlers
  416.                 int byteRead = performRead();
  417.             
  418.                 // stream completed
  419.                 if(byteRead == -1){
  420.                     //System.out.println("@@@@@@@@ IS complite SINGLE BYTE ");
  421.                     this.isOriginalInputStreamConsumed = true;
  422.                 }
  423.                 
  424.                 // if enabled buffering, the byte read is saved
  425.                 if(this.bufferEnabled){
  426.                     if(this.isOriginalInputStreamConsumed){
  427.                         //System.out.println("@@@@@@@@ IS finalize! SINGLE BYTE");
  428.                         this.finalizeBuffer();
  429.                     }else{
  430.                         //System.out.println("@@@@@@@@ write SINGLE BYTE["+byteRead+"] ");
  431.                         this.buffer.write(byteRead);
  432.                     }
  433.                 }
  434.                 
  435.                 //System.out.println("@@@@@@@@ Return dopo Perform SINGLE BYTE "+byteRead);
  436. //              if(byteRead == -1){
  437. //                  this.printFile();
  438. //              }else{
  439. //                  this.debug(new byte[]{(byte)byteRead}, 0, 1);
  440. //              }
  441.                 return byteRead;
  442.             }
  443.         }finally{
  444.             if(incrementCurrentReadPositionFromStream){
  445.                 this.currentReadPositionFromStream++;
  446.             }
  447.         }
  448.     }
  449.     
  450.     private int performRead() throws IOException {
  451.         int b = this.is.read();

  453.         // enable dispatching to streaming handlers
  454.         for(String streamingHandlerId : this.streamingHandlersIds) {
  455.             StreamingHandler streamingHandler = this.streamingHandlers.get(streamingHandlerId);
  456.             try{
  457.                 if(b==-1){
  458.                     //System.out.println("@@@@@@@@ return -1 Dispatching end ...");
  459.                     streamingHandler.end();
  460.                 }
  461.                 else{
  462.                     //System.out.println("@@@@@@@@ return un byte Dispatching ...");
  463.                     streamingHandler.feed((byte)b);
  464.                 }
  465.             }catch(Throwable e){
  466.                 if(this.log!=null){
  467.                     this.log.error("["+streamingHandlerId+"] error occurs: "+e.getMessage(),e);
  468.                 }
  469.                 if(this.throwStreamingHandlerException){
  470.                     throw new IOException("["+streamingHandlerId+"] "+e.getMessage());
  471.                 }
  472.             }
  473.         }
  474.         return b;
  475.     }
  476.     
  477.     @Override
  478.     public void close() throws IOException {
  479.         
  480.         if(this.isClosed==false){
  481.         
  482.             if(this.is!=null)
  483.                 this.is.close();
  484.             
  485.             IOException streamingHandlerException = null;
  486.             
  487.             // enable dispatching to streaming handlers
  488.             for(String streamingHandlerId : this.streamingHandlersIds) {
  489.                 StreamingHandler streamingHandler = this.streamingHandlers.get(streamingHandlerId);
  490.                 try{
  491.                     streamingHandler.closeResources();
  492.                 }catch(Throwable e){
  493.                     if(this.log!=null){
  494.                         this.log.error("["+streamingHandlerId+"] error occurs: "+e.getMessage(),e);
  495.                     }
  496.                     if(this.throwStreamingHandlerException){
  497.                         if(streamingHandlerException==null){
  498.                             // throw the first exception occurs
  499.                             streamingHandlerException = new IOException("["+streamingHandlerId+"] "+e.getMessage());
  500.                         }
  501.                     }
  502.                 }
  503.             }
  504.                     
  505.             this.isClosed = true;
  506.             
  507.             if(streamingHandlerException!=null){
  508.                 throw streamingHandlerException;
  509.             }
  510.         }
  511.     }
  512.     
  513.     
  514.     
  515.     
  516.     
  517.     
  518.     /* ******* HANDLERS ******* */
  519.     
  520.     /**
  521.      * adds a Streaming handler to our list
  522.      * @param handler handler to add
  523.      * @throws IOException 
  524.      */
  525.     public void addStreamingHandler(StreamingHandler handler) throws IOException {
  526.         
  527.         //System.out.println("@@@@@@@@ addStreamingHandler currentReadPositionFromStream["+this.currentReadPositionFromStream+"] COMPLITED["+this.isCompleted+"]... ");
  528.         
  529.         // the option of buffering can be enabled only if the stream has not yet been accessed
  530.         if(this.currentReadPositionFromStream>0){
  531.             if(this.bufferEnabled==false)
  532.                 throw new IOException("You can not add handler after the stream has been accessed with buffering disabled");
  533.             else {
  534.                 
  535.                 String id = handler.getID();
  536.                 if(this.streamingHandlersIds.contains(id)){
  537.                     throw new IOException("StreamingHandler with id ["+id+"] already exists");
  538.                 }
  539.                 this.streamingHandlers.put(id, handler);
  540.                 this.streamingHandlersIds.add(id);
  541.                                 
  542.                 if(this.isOriginalInputStreamConsumed){
  543.                     //System.out.println("@@@@@@@@ addStreamingHandler FEED ["+this.contentCompleteReadedFromStream.length+"]");
  544.                     handler.feed(this.contentCompleteReadedFromStream);
  545.                     
  546.                     //System.out.println("@@@@@@@@ addStreamingHandler END");
  547.                     handler.end();
  548.                 }
  549.                 else{
  550.                     //System.out.println("@@@@@@@@ addStreamingHandler FEED ["+this.buffer.size()+"]");
  551.                     handler.feed(this.buffer.toByteArray());
  552.                 }
  553.             }
  554.         }
  555.         
  556.     }
  557.     
  558.     /**
  559.      * Given a StreamingHandler class, returns the corresponding handler, if one
  560.      * Useful in case of more handlers registered to this class
  561.      * @param clazz - the class which the needed handler belongs
  562.      * @return corresponding handler if there is one, otherwise null
  563.      * @throws IOException 
  564.      */
  565.     @SuppressWarnings("unchecked")
  566.     public <T extends StreamingHandler> T getFirstStreamingHandlerByType(Class<T> clazz) throws IOException {
  567.         for(String streamingHandlerId : this.streamingHandlersIds) {
  568.             StreamingHandler streamingHandler = this.streamingHandlers.get(streamingHandlerId);
  569.             if (streamingHandler.getClass().equals(clazz)) 
  570.                 return (T) streamingHandler;
  571.         }
  572.         throw new IOException("StreamingHandler with type ["+clazz.getName()+"] not exists");
  573.     }

  575.     public StreamingHandler getStreamingHandler(String id) throws IOException{
  576.         if(this.streamingHandlersIds.contains(id)){
  577.             return this.streamingHandlers.get(id);
  578.         }
  579.         else{
  580.             throw new IOException("StreamingHandler with id ["+id+"] not exists");
  581.         }
  582.     }
  583.     

  585.     
  586.     

  588.     
  589.     
  590.     /* ******* BUFFER ******* */
  591.     
  592.     public void setONBuffering() throws IOException {
  593.         
  594.         // the option of buffering can be enabled only if the stream has not yet been accessed
  595.         if(this.currentReadPositionFromStream>0){
  596.             throw new IOException("You can not enable buffering after the stream has been accessed");
  597.         }

  599.         if(this.bufferEnabled){
  600.             throw new IOException("The buffering is already enabled");
  601.         }
  602.         
  603.         // enable buffering
  604.         this.bufferEnabled = true;
  605.         this.buffer = new ByteArrayOutputStream();

  607.     }
  608.     
  609.     public void setOFFBuffering() throws IOException {
  610.         this.setOFFBuffering(true);
  611.     }
  612.     public void setOFFBuffering(boolean releaseBufferReaded) throws IOException {
  613.         
  614.         //System.out.println("@@@@@@@@ setOFFBuffering("+releaseBufferReaded+")");
  615.         
  616.         if(this.bufferEnabled==false){
  617.             throw new IOException("The buffering is not enabled");
  618.         }
  619.         
  620.         // disable buffer
  621.         this.bufferEnabled = false;
  622.         if(releaseBufferReaded){
  623.             //System.out.println("@@@@@@@@ setOFFBuffering("+releaseBufferReaded+") rilascio");
  624.             this.contentCompleteReadedFromStream = null;
  625.             if(this.buffer!=null){
  626.                 this.buffer.close();
  627.                 this.buffer = null;
  628.                 //System.out.println("@@@@@@@@ setOFFBuffering("+releaseBufferReaded+") rilasciato");
  629.             }
  630.         }

  632.     }
  633.     
  634.     private void finalizeBuffer() throws IOException{
  635.         
  636.         if(this.bufferEnabled==false){
  637.             throw new IOException("BufferMode is not enabled");
  638.         }
  639.         
  640.         if(this.contentCompleteReadedFromStream!=null){
  641.             throw new IOException("bufferingComplete already invoked");
  642.         }
  643.         
  644.         this.buffer.flush();
  645.         this.buffer.close();
  646.         this.contentCompleteReadedFromStream=this.buffer.toByteArray();
  647.         this.buffer=null; // G.C.
  648.         
  649.     }
  650.     
  651.     
  652.     
  653.     
  654.     
  655.     
  656.     
  657.     
  658.     
  659.     
  660.     /* ******* SERIALIZE ******* */
  661.     
  662.     public void serialize(OutputStream out) throws IOException{
  663.         this.serializeEngine(out,false);
  664.     }
  665.     
  666.     public void serializeAndConsume(OutputStream out) throws IOException{
  667.         this.serializeEngine(out,true);
  668.     }
  669.     
  670.     public byte[] serialize() throws IOException{
  671.         ByteArrayOutputStream bout = new ByteArrayOutputStream();
  672.         this.serializeEngine(bout,false);
  673.         bout.flush();
  674.         bout.close();
  675.         return bout.toByteArray();
  676.     }
  677.     
  678.     public byte[] serializeAndConsume() throws IOException{
  679.         ByteArrayOutputStream bout = new ByteArrayOutputStream();
  680.         this.serializeEngine(bout,true);
  681.         bout.flush();
  682.         bout.close();
  683.         return bout.toByteArray();
  684.     }
  685.     
  686.     private void serializeEngine(OutputStream out,boolean consume) throws IOException{
  687.         
  688.         //System.out.println("@@@@@@@@ serializeEngine");
  689.         
  690.         // If the stream is already saved return byte array
  691.         //if(this.bufferEnabled && this.contentCompleteReadedFromStream!=null){
  692.         if(this.contentCompleteReadedFromStream!=null){
  693.             out.write(this.contentCompleteReadedFromStream);
  694.             return;
  695.         }
  696.         
  697.         // If buffering is enabled, but the stream is not completely consumed, write the bytes stored in the buffer in the output stream
  698.         //if(this.bufferEnabled && this.contentCompleteReadedFromStream==null){
  699.         if(this.buffer!=null){
  700.             this.buffer.flush();
  701.             if(this.buffer.size()>0){
  702.                 out.write(this.buffer.toByteArray());
  703.             }
  704.         }
  705.         
  706.         if(consume){
  707.             if(this.bufferEnabled){
  708.                 this.setOFFBuffering();
  709.             }
  710.             else if(this.buffer!=null){
  711.                 this.buffer.close();
  712.                 this.buffer = null;
  713.             }
  714.         }
  715.         
  716.         // Conclude to consume the stream remaining
  717.         // If buffering is enabled all bytes read will stored in the buffer
  718.         byte[] buffer = new byte[1024];
  719.         int byteRead = this.read_engine(buffer, 0, buffer.length, false);
  720.         while ( byteRead != -1 ){
  721.             out.write(buffer,0,byteRead);
  722.             byteRead = this.read_engine(buffer, 0, buffer.length, false);
  723.         }
  724.     }
  725.     

  727.     
  728.     

  730. }
Created with JaCoCo 0.8.8.202204050719