mqsload/MQSLoad.java

// $RCSfile: MQSLoad.java,v $
// $Revision: 1.3 $
// $Name:  $
// $Date: 2002/04/19 15:22:14 $
// $Author: giberta1 $

/*
 * MQSLoad.java - Data file to MQ/Series queue loader
 * Copyright (C) 2001-2002 Arnaud G. Gibert
 * arnaud.gibert@misys.com
 * www.miys-ibs.fr
 *
 * 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 com.ibm.mq.*;
import java.io.*;





public class MQSLoad
{
        private MQQueueManager  QMng;
        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 String          File_Name;
        private String		Field_Break = "";
        private String		Message_Break = "\n";
        




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

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

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

        
        
        
        
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
	private void Arg_Parse( String  args[]) throws Exception
        {
        	int  argc = 0;
        	
        	
        	try
                {
			while( argc < args.length)
                        {
                		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
                		{
        	                        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
        {
                try
                {
                        QMng = new MQQueueManager( QMng_Name);
                        System.out.println( "QManager Open: (" + QMng + ") !");
                }
                
                catch( Exception Expt)
                {                       
                        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( int Msg_Nb)
	{
		if( ( Msg_Nb % 50) == 0)
		{
			System.out.print( "\n(" + Msg_Nb + ")\t");
		}

		if( ( Msg_Nb % 10) == 0)
		{
			System.out.print( " ");
		}
		
		System.out.print( ".");
	}
	
	
	
	
	
	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        private void Load_File( BufferedInputStream  Input_File) throws Exception
        {
        	byte       input_char;
                int        msg_nb = 0;
                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;
                byte[]     buffer_byte;
                String     buffer_string, out_string = null;
                
                
                try
                {
	                output_msg.format = MQC.MQFMT_STRING;
	                
	                
	                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 > 0)
				{
					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;
					
					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);
						}
						
						if( next_message <= cur_window_size)
						{
//							System.out.println( "* Write Msg CurPos: (" + cur_pos + ")");
		                                        output_msg.messageId = MQC.MQMI_NONE;
		                                        MQSPut_Msg( output_msg);
		                                        output_msg.clearMessage();
		                                        out_string = null;
		                                        cur_pos += Message_Break.length();
		                                        
		                                        Counter_Print( msg_nb++);
		                		}
					}
					
					if( cur_buffer_size > cur_window_size)
					{
						buffer_offset = cur_buffer_size - cur_window_size;
						System.arraycopy( buffer_byte, cur_window_size, buffer_byte, 0, buffer_offset);
					}
				}
			}
			while( read_size > 0);

			if( out_string != null)
			{
//				System.out.println( "* Flushing Msg");
                                output_msg.messageId = MQC.MQMI_NONE;
                                MQSPut_Msg( output_msg);
                                
                                Counter_Print( msg_nb++);
        		}

	                System.out.println( "\n");
	   	}
	   	
                catch( Exception Expt)
                {
                        throw Expt;
                }
        }
	   	

		                                


	//-------------------------------------------------------------------------------------------------------------------------
	//
	//-------------------------------------------------------------------------------------------------------------------------
	
        public MQSLoad( String args[])
        {
		try
		{
                                      
                        if( ( args.length < 2) || ( args.length > 6))
                        {
                                System.out.println( "Usage: MQSLoad [-fb \"field_break\"] [-mb \"message_break\"] <Output_MsgQueue_Name> <Input_File_Name>");
                        }
                        else
                        {
                                System.out.println( "MQS Load Starting...");
                                
                                Arg_Parse( args);
                    
                                System.out.println( "Output MsgQueue Name: (" + MsgQ_Name + ")   Input File Name: (" + File_Name + ")");
                                System.out.println( "Field Break: (" + Field_Break + ")   Message_Break: (" + Message_Break + ")");
                
                                MQSInit();

				try
				{	
	                                BufferedInputStream input_file = new BufferedInputStream( new FileInputStream( File_Name));
	                                
	                                try
	                                {
		                                System.out.println("Input File Open:  (" + input_file + ") !");
		                                
		                                Load_File( input_file);
		                                                
		                        }
		                        
		                        catch( Exception Expt)
		                        {
 	        	                        input_file.close();
		        	                throw Expt;
	                       		}

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

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