mqsload/MQSLoad.java

// $RCSfile: MQSLoad.java,v $
// $Revision: 1.15 $
// $Name:  $
// $Date: 2008/12/04 15:02:16 $
// $Author: agibert $

/*
 * MQSLoad.java - Data file to MQ/Series queue loader
 * Copyright (C) 2001-2006 Arnaud G. Gibert
 * mailto:arnaud@rx3.net
 * http://www.rx3.org/dvp/MQSLoad
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or any later version.
 *
 * 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, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */

import java.io.*;
import java.text.*;
import com.ibm.mq.*;





public class MQSLoad
{
	private String		MQSLoad_Revision   = "$Revision: 1.15 $";
	private String    	MQSLoad_Tag        = "$Name:  $";
	private	String    	MQSLoad_Date       = "$Date: 2008/12/04 15:02:16 $";
	private	String 		MQSLoad_Author     = "$Author: agibert $";
	private MQQueueManager  QMng               = null;
	private String          QMng_Name          = "";
	private MQQueue         MsgQ;
	private String          MsgQ_Name;
	private int             MsgQ_Open_Options  = MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING;
	private int		Sleep_Time 	   = 0;
	private String		Field_Break        = "";
	private String		Message_Break      = "\r\n";
	private String		Message_Tail       = "";
	private int		Message_Skip	   = 0;
	private int		Message_Count	   = 0;
	private int		Repeat_Count	   = 1;
	private boolean		Keep_Message_Break = false;
	private String          File_Name;
	private int             Msg_Counter        = 0;





	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        public static void  main( String  args[])
        {
	    new MQSLoad( args);
        }

        
        
        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        public static String  Str_Format( String  UnFmt)
        {
	    int	 idx;
	    boolean  esc = false;
	    String   fmt = "";
        	
        	
	    for( idx = 0; idx < UnFmt.length(); idx++)
	    {
		if( esc)
		{
		    switch( UnFmt.charAt( idx))
		    {
			case '\\':
			{
			    fmt += "\\";
			    break;
			}

			case 'n':
			{
			    fmt += "\n";
			    break;
			}
                			
			case 'r':
			{
			    fmt += "\r";
			    break;
			}
                			
			case 't':
			{
			    fmt += "\t";
			    break;
			}
                			
			default:
			{
			    fmt += "?";
			    break;
			}             		
		    }
                		
		    esc = false;
		}
		else
		{
		    if( UnFmt.charAt( idx) == '\\')
		    {
			esc = true;
		    }
		    else
		    {
			fmt += UnFmt.charAt( idx);
		    }
		}
	    }
                
	    return( fmt);
        }

        
        
        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        public static String  Str_UnFormat( String  Fmt)
        {
	    int	 idx;
	    String   unfmt = "";
        	
        	
	    for( idx = 0; idx < Fmt.length(); idx++)
	    {
		switch( Fmt.charAt( idx))
		{
		    case '\\':
		    {
			unfmt += "\\\\";
			break;
		    }

		    case '\n':
		    {
			unfmt += "\\n";
			break;
		    }
        			
		    case '\r':
		    {
			unfmt += "\\r";
			break;
		    }
        			
		    case '\t':
		    {
			unfmt += "\\t";
			break;
		    }
        			
		    default:
		    {
			unfmt += Fmt.charAt( idx);
			break;
		    }             		
		}
	    }
                
	    return( unfmt);
        }

        
        
        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
	private void  Arg_Parse( String  args[]) throws Exception
        {
	    int  argc = 0;
        	
        	
	    try
	    {
		while( argc < args.length)
		{
		    if( args[argc].equals( "-qm"))
		    {
			if( argc < ( args.length + 1))
			{
			    QMng_Name = Str_Format( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if( args[argc].equals( "-fb"))
		    {
			if( argc < ( args.length + 1))
			{
			    Field_Break = Str_Format( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-mb"))
		    {
			if( argc < ( args.length + 1))
			{
			    Message_Break = Str_Format( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-mt"))
		    {
			if( argc < ( args.length + 1))
			{
			    Message_Tail = Str_Format( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-st"))
		    {
			if( argc < ( args.length + 1))
			{
			    Sleep_Time = Integer.parseInt( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-ms"))
		    {
			if( argc < ( args.length + 1))
			{
			    Message_Skip = Integer.parseInt( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-mc"))
		    {
			if( argc < ( args.length + 1))
			{
			    Message_Count = Integer.parseInt( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-rc"))
		    {
			if( argc < ( args.length + 1))
			{
			    Repeat_Count = Integer.parseInt( args[++argc]);
			}
			else
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
		    else if ( args[argc].equals( "-kmb"))
		    {
			Keep_Message_Break = true;
		    }
		    else
		    {
			MsgQ_Name = args[argc++];
			File_Name = args[argc++];
                			
			if( argc < args.length)
			{
			    System.out.println( "Invalid number of command line options...");
			    throw new Exception();
			}
		    }
                		
		    ++argc;
		}
	    }
                
	    catch( Exception Expt)
	    {                       
		throw Expt;
	    }
	}
        
                
                
                
                       
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
      	private void  MQSInit() throws Exception
        {
	    short  	     retry = 0;
	    final short  RETRY_MAX = 10;


	    while( QMng == null)
	    {
		try
		{
		    QMng = new MQQueueManager( QMng_Name);
//                        	System.out.println( "QManager Open: (" + QMng + ") !");
		}
                
		catch( Exception Expt)
		{
		    System.out.print("!");

		    if( retry++ > RETRY_MAX)
		    {
			throw Expt;
		    }
		}
	    }
                
	    try
	    {
		MsgQ = QMng.accessQueue( MsgQ_Name, MsgQ_Open_Options, null, null, null);
//                        System.out.println( "MsgQ Open:  (" + MsgQ + ") !");
	    }
   
	    catch( Exception Expt)
	    {      
		QMng.disconnect();
//                        System.out.println( "QManager Close:  (" + QMng + ") !");                        

		throw Expt;
	    }
        }



        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private void  MQSDeInit() throws Exception
        {
	    try
	    {
		MsgQ.close();
//                        System.out.println( "MsgQ Close:  (" + MsgQ + ") !");
		QMng.disconnect();
//                        System.out.println( "QManager Close:  (" + QMng + ") !");
	    }

	    catch( Exception Expt)
	    {
		throw Expt;
	    }
        }

        
        
        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private void  MQSPut_Msg( MQMessage Msg) throws Exception
        {
	    try
	    {
		MQPutMessageOptions pmo = new MQPutMessageOptions();
		MsgQ.put( Msg, pmo);
	    }
                
	    catch( Exception Expt)
	    {
		throw Expt;
	    }
        }

        


        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
	private void  Counter_Print( String  Mark)
	{
	    if( ( Msg_Counter % 50) == 0)
	    {
		System.out.print( "\n(" + Msg_Counter + ")\t");
	    }

	    if( ( Msg_Counter % 10) == 0)
	    {
		System.out.print( " ");
	    }

	    System.out.print( Mark);

	    Msg_Counter++;
	}
	
	
	
	
	
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------

	private void  Stand_By(int timer) throws Exception
	{
	    if( timer > 0)
	    {
		try
		{
		    synchronized( this)
		    { 
			wait( timer);				
		    }
		}
		catch(InterruptedException  ie)
		{
		    ie.printStackTrace();
		}
	    }
	}





	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private int  Put_Msg( MQMessage  Output_Msg, int  Msg_Id, int  Msg_Nb, int  Msg_Skip, int  Msg_Count) throws Exception
	{
	    if( Msg_Id < Msg_Skip)
	    {
		Counter_Print( "*");
	    }
	    else
	    {
		if( ( Msg_Count != 0) && ( Msg_Id >= ( Msg_Skip + Msg_Count)))
		{
		    Counter_Print( "#");
		}
		else
		{
		    Output_Msg.messageId = MQC.MQMI_NONE;
		    MQSPut_Msg( Output_Msg);
		    Output_Msg.clearMessage();
		    
		    Counter_Print( ".");
		    
		    Msg_Nb++;

		    Stand_By( Sleep_Time);
		}
	    }

	    return(Msg_Nb);
	}





	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private int  Load_File( RandomAccessFile  Input_File, int  Msg_Nb, int  Msg_Skip, int  Msg_Count) throws Exception
        {
	    byte       input_char;
	    MQMessage  output_msg = new MQMessage();
	    int	       buffer_size, cur_buffer_size, window_size, cur_window_size, prefetch_size, buffer_offset, read_size;
	    int	       cur_pos, end_pos, next_message, next_field;
	    int        msg_id = 0;
	    byte[]     buffer_byte;
	    String     buffer_string, out_string = null;
                
                
	    try
	    {
		output_msg.format = MQC.MQFMT_STRING;
	                
	                
		/* Default window size */
		window_size = 1024 * 64;
		prefetch_size = Field_Break.length() + Message_Break.length();
		buffer_size = window_size + prefetch_size;
	                
		buffer_offset = 0;
	               
		buffer_byte = new byte[buffer_size];
	                
		do
		{		
		    read_size = Input_File.read( buffer_byte, buffer_offset, buffer_size - buffer_offset);
//				System.out.println( "Load buffer read_size: (" + read_size + ")   asked: (" + (buffer_size - buffer_offset) + ")");
				
		    if( read_size == -1)
		    {
			read_size = 0;
		    }

		    /* Compute current buffer and window size */
		    cur_window_size = Math.min( ( read_size + buffer_offset), window_size);
		    cur_buffer_size = ( read_size + buffer_offset);

		    buffer_string = new String( buffer_byte, 0, cur_buffer_size);
		    cur_pos = 0;
					
//				System.out.println( "  CurWS: (" + cur_window_size + ")   CurBS: (" + cur_buffer_size + ")   Buffer: [" + buffer_string + "]");

		    while( cur_pos < cur_window_size)
		    {
//				        System.out.println( "- CurPos: (" + cur_pos + ")");
			next_message = buffer_string.indexOf( Message_Break, cur_pos);
							
			if( ( next_message == -1) || ( next_message > cur_window_size))
			{
			    next_message = cur_window_size + 1;
			    end_pos = cur_window_size;
			}
			else
			{
//					        System.out.println( "+ Get message   next_message: (" + next_message + ")");
			    end_pos = next_message;
			}
	
			while( cur_pos < end_pos)
			{
//					        System.out.println( "+ CurPos: (" + cur_pos + ")   end_pos: (" + end_pos + ")");
			    if( Field_Break.length() > 0)
			    {
				next_field = buffer_string.indexOf( Field_Break, cur_pos);
			    }
			    else
			    {
				next_field = -1;
			    }
							
			    if( ( next_field == -1) || ( next_field > end_pos))
			    {
				out_string = buffer_string.substring( cur_pos, end_pos);
				cur_pos = end_pos;
			    }
			    else
			    {
//						        System.out.println( "+ Get field   next_field: (" + next_field + ")");
				out_string = buffer_string.substring( cur_pos, next_field);
				cur_pos = next_field + Field_Break.length();
			    }
								
			    output_msg.writeString( out_string);
//						System.out.println( "* Msg: [" + out_string + "]");
						
			}
						
			if( next_message <= cur_window_size)
			{
			    /* A message break has been found: send the data... */
//			    System.out.println( "* Write Msg   CurPos: (" + cur_pos + ")");
			    output_msg.writeString( Message_Tail);

			    if( Keep_Message_Break)
			    {
				output_msg.writeString( Message_Break);
			    }

//			    System.out.println( "(" + Msg_Nb + ")");
			    Msg_Nb = Put_Msg( output_msg, msg_id, Msg_Nb, Msg_Skip, Msg_Count);
			    
			    msg_id++;
			    out_string = null;
			    cur_pos += Message_Break.length();
			}
		    }
				
		    buffer_offset = cur_buffer_size - cur_pos;
//				System.out.println( "% CurPos: (" + cur_pos + ")   BufOff: (" + buffer_offset + ")");

		    if( buffer_offset > 0)
		    {
			/* Copy back prefetch area */
			System.arraycopy( buffer_byte, cur_pos, buffer_byte, 0, buffer_offset);
		    }
		}
		while( cur_buffer_size > 0);

		if( out_string != null)
		{
//		    System.out.println( "* Flushing Msg");
		    Msg_Nb = Put_Msg( output_msg, msg_id, Msg_Nb, Msg_Skip, Msg_Count);
		}

		return( Msg_Nb);
	    }
	   	
	    catch( Exception Expt)
	    {
		throw Expt;
	    }
        }
	   	

		                                


	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private void  Print_Usage( ) throws Exception
        {
	    System.out.println( "Usage: MQSLoad [-qm \"Output_QueueMng_Name\"] [-st \"sleep_time\"] [-fb \"field_break\"] [-mb \"message_break\"] [-mt \"message_tail\"] [-kmb] [-ms \"message_skip\"] [-mc \"message_count\"] [-rc \"repeat_count\"] <Output_MsgQueue_Name> <Input_File_Name>");
	    System.out.println( "       Default: Output QueueMng Name: (" + QMng_Name + ")   Sleep Time: (" + Sleep_Time + ")   Field Break: (" + Str_UnFormat( Field_Break) + ")   Message Break: (" + Str_UnFormat( Message_Break) + ")   Message Tail: (" + Str_UnFormat( Message_Tail) + ")   Keep Message Break: (" + Keep_Message_Break + ")   Message Skip: (" + Message_Skip + ")   Message Count: (" + Message_Count + ")   Repeat Count: (" + Repeat_Count + ")");
	}





	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private void  Print_Args( ) throws Exception
        {
	    System.out.println( "Output QueueMng Name: (" + QMng_Name + ")   Output MsgQueue Name: (" + MsgQ_Name + ")   Input File Name: (" + File_Name + ")");
	    System.out.println( "Sleep Time: (" + Sleep_Time + ")   Field Break: (" + Str_UnFormat( Field_Break) + ")   Message Break: (" + Str_UnFormat( Message_Break) + ")   Message Tail: (" + Str_UnFormat( Message_Tail) + ")   Keep Message Break: (" + Keep_Message_Break + ")   Message Skip: (" + Message_Skip + ")   Message Count: (" + Message_Count + ")   Repeat Count: (" + Repeat_Count + ")");
	}

                



	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
	public MQSLoad( String args[])
	{
	    int           msg_nb               = 0;
	    int           loop;
	    long          time_begin, time_end;
	    double        time_elapsed, speed;
	    NumberFormat  nf                   = NumberFormat.getInstance();
		
		
	    try
	    {
		System.out.println("MQSLoad: " + MQSLoad_Tag + " / " + MQSLoad_Date + " / " + MQSLoad_Author);
                                      
		if( ( args.length < 2) || ( args.length > 19))
		{
		    Print_Usage();
		    System.exit( 1);
		}
		else
		{
		    System.out.println( "MQS Load Starting...");
                                
		    Arg_Parse( args);

		    Print_Args();
                    
		    MQSInit();

		    try
		    {	
			RandomAccessFile input_file = new RandomAccessFile( File_Name, "r");
	                                
			try
			{
//		                                System.out.println("Input File Open:  (" + input_file + ") !");

			    time_begin = System.currentTimeMillis();

			    for( loop = 0; loop < Repeat_Count; loop++ )
			    {
				try
				{
				    input_file.seek(0);
				}

				catch( Exception  Expt)
				{
				    if( Repeat_Count > 1)
				    {
					System.out.println( "Repeat Count should be 1 for non seekable file !");
					throw Expt;
				    }
				}
				
				msg_nb = Load_File( input_file, msg_nb, Message_Skip, Message_Count);
			    }

			    time_end = System.currentTimeMillis();
		                                
			    time_elapsed = ( time_end - time_begin) / 1000.0;
			    speed = msg_nb / time_elapsed;
		                               	
			    nf.setMinimumFractionDigits(2);
			    nf.setMaximumFractionDigits(2);
		                               	
			    System.out.println( "\n");
			    System.out.println( "Loaded Message Nb: (" + msg_nb
						+ ")   Elapsed Time: (" + nf.format(time_elapsed)
						+ ") s   Speed: (" + nf.format(speed)
						+ ") msg/s");
			}
		                        
			catch( Exception Expt)
			{
			    input_file.close();
			    throw Expt;
			}

			input_file.close();
		    }
	                        
		    catch( Exception Expt)
		    {
			MQSDeInit();
			throw Expt;
		    }
                        
		    MQSDeInit();
		    System.out.println( "MQS Load Completed !");

		    System.exit( 0);
		}
	    }

	    catch( Exception Expt)
	    {
		System.out.println("Exception:  (" + Expt + ") !");
		Expt.printStackTrace();
                        
		System.exit( 1);
	    }
	}



	//-------------------------------------------------------------------------------------------------------------------------

}