bocomq.pm

Perl开发的数据采集工具 基于UNIX

#******************************************************************
#File:		CAQMessagingQueue.pm
#Author:	Chenggen Dong<[e-mail]>
#Date:		May 7 2002
#Description:
#			This is a perl library for receiving/sending message
#			through IBM MQSeries.
#			And this library dependent on MQSerice Client library
#			for perl.
#			This API can send message which is no more than 32767
#			bytes,and getMsgLength() will not return the true msg
#			length,while return the buffer length.
#Tested against:
#			Sun Solaris 2.x
#
#Copyright @ Bright Ocean Inter-Telecom
#All rights reserved
#
#Revision:
#
#******************************************************************
use strict;
use MQClient::MQSeries qw(:functions);
use MQSeries::Queue;
use MQSeries::Message;

package BocoMQ;

use Data::Dumper;

###################################################################
#	constant definition
###################################################################
my $MANAGER_OK					= 0;
my $QUEUE_OK					= 0;
my $MESSAGE_OK					= 0;
my $MESSAGE_MAX_LENGTH				= 2048;
my $PRIORITY_ZERO				= 0;
my $MSG_TYPE_FIRST				= 65536;
my $MSG_TYPE_LAST				= 999999999;
my $MSG_TYPE_REAL				= $MSG_TYPE_LAST-$MSG_TYPE_FIRST;
my $WAIT_FOREVER				= -1;
my $DEFAULT_MANAGER_NAME			= "DEFAULT_QM";
my $USE_DEFAULT_PRIORITY			= -1;

my $MANAGER_ERROR				= -31100;
my $MANAGER_NOT_CONNECTED			= -31101;
my $MANAGER_QUEUE_NOT_MATCH			= -31102;
my $MANAGER_ALREADY_CONNECTED			= -31103;
my $MANAGER_DISC_ERROR				= -31104;
my $MANAGER_NOT_OK				= -31105;

my $QUEUE_ERROR					= -31200;
my $QUEUE_NOT_INITIALISED			= -31201;
my $QUEUE_ALREADY_INITIALISED			= -31203;
my $QUEUE_OPEN_ERROR				= -31205;
my $QUEUE_INIT_ERROR				= -31206;

my $MESSAGE_ERROR				= -31300;
my $MESSAGE_SEND_ERROR				= -31301;
my $MESSAGE_SEND_INVALID			= -31302;
my $MESSAGE_SEND_NOTALLOWED			= -31303;
my $MESSAGE_SEND_QUEUE_FULL			= -31304;
my $MESSAGE_RECEIVE_NONE			= -31305;
my $MESSAGE_RECEIVE_ERROR			= -31306;
my $MESSAGE_RECEIVE_INVALID			= -31307;
my $MESSAGE_TYPE_ERROR				= -31309;
my $MESSAGE_TIMEOUT_ERROR			= -31310;
###################################################################
#	end constant definition
###################################################################


###################################################################
#	class variants definition
###################################################################
#my $initialised					= 0;
#my $complete_code				= 0;
#my $reason_code					= 0;
#my $m_name						= "";
#my $q_name						= "";
#my $msg_type					= -1;
#my $msg_priority				= -1;
#my $msg_length					= -1;
###################################################################
#	end class variants definition
###################################################################

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name:	new($qname,$mname)
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				Constructor of CAQMessagingQueue class
#Parameters:
#	1.	$qname:queue name for receiving/sending message,no default
#	2.	$mname:manager name for receiving/sending message,default is DEFAULT_QM
#Return: self reference
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub new
{
	my $class = shift;
	my %hash = @_;
	my $self = {};
	bless $self,$class;
	#start to initialize class variants
	$self->{m_name} = $hash{mname} || "DEFAULT_QM";
	$self->{q_name} = $hash{qname} || "TEST.Q";
	$self->{initialised} = 0;
	$self->{queue} = 0;
	$self->{complete_code} = 0;
	$self->{reason_code} = 0;
	$self->{msg_type} = -1;
	$self->{err_msg} = "";
	$self->{msg_priority} = -1;
	$self->{msg_length} = -1;
	#end initialization
	return $self;
}#new

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: DESTROY()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				Deconstructor of CAQMessagingQueue class
#Parameters:
#				None
#Return:
#				None
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub DESTROY
{
	my $self = shift;
	my $options = {};
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: initialize()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				initialize,connect manager and open queue
#Parameters:
#				None
#Return:
#				success:0
#				fail:none 0
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub initialize
{
	my $self = shift;
	$self->{queue} = MQSeries::Queue->new
		(
			QueueManager	=> $self->{m_name},
			Queue			=> $self->{q_name},
			Options			=> MQSeries::MQOO_INQUIRE | MQClient::MQSeries::MQOO_OUTPUT | MQClient::MQSeries::MQOO_INPUT_AS_Q_DEF | MQSeries::MQOO_SET,
		)
		or die("Unable to open queue.\n");
		return 0;
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: showError()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				return err_msg
#Parameters:
#				None
#Return:
#				$self->{err_msg}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getErrorMsg
{
	my $self = shift;
	if( $self->{complete_code} != $MANAGER_OK || $self->{reason_code} != MQClient::MQSeries::MQRC_NONE )
	{
		return ($self->{err_msg});
	}
	else
	{
		$self->{err_msg} = "NO ERROR";
		return ($self->{err_msg});
	}
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getReasonCode()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current reason code
#Parameters:
#				None
#Return:
#				$self->{reason_code}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getReasonCode
{
	my $self = shift;
	return ($self->{reason_code});
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getCompleteCode()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current complete code
#Parameters:
#				None
#Return:
#				$self->{complete_code}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getCompleteCode
{
	my $self = shift;
	return ($self->{complete_code});
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: sendMsg()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				send message through IBM MQ
#Parameters:
#	1: msgBuffer,which stores the send message,required
#	2: msgType,which is the message type,required
#	3: msgPriority,which is the message priority,optional,default is 0.the lowest
#	4: msgExpiry,which is the expiry time of message,optional,default is -1(alive forever)
#Return:
#				0:success
#				None 0:fail
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub sendMsg
{
	my $self = shift;
	#my %hash = @_;
	#my $buffer = $hash{msgBuffer};
	#my $type = $hash{msgType};
	my ($buffer,$type,$priority,$expiry) = @_;
	if( !defined($type) )
	{
		return $MESSAGE_TYPE_ERROR;
	}
	if($type < 0 || $type >= $MSG_TYPE_REAL)
	{
		return $MESSAGE_TYPE_ERROR;
	}
	$type += $MSG_TYPE_FIRST;
	if( !defined($priority) )
	{
		$priority = 0;
	}
	if( $priority > 8 || $priority < 0)
	{
		$priority = 0;
	}
	if( !defined($expiry) )
	{
		$expiry = -1;
	}

	my %hash_msg_desc = {
		#MsgType			=> $type,
		#Priority		=> $priority,
		#Format			=> MQClient::MQSeries::MQFMT_NONE,
		#Expiry			=> $expiry,
	};
	my $msg_desc = \%hash_msg_desc;
	my $put_msg = MQSeries::Message->new
		(
			MsgDesc		=>
			{
				MsgType		=> $type,
				Priority	=> $priority,
				Format		=> MQClient::MQSeries::MQFMT_NONE,
				Expiry		=> $expiry,
			},
			Data		=> $buffer,
		);
print Dumper($put_msg);exit;
	my $return_value = $self->{queue}->Put( Message => $put_msg );
	if( $return_value == 0 )
	{
	}
	$self->{complete_code} = $self->{queue}->CompCode();
	$self->{reason_code} = $self->{queue}->Reason();
	if( $self->{complete_code} != $MANAGER_OK || $self->{reason_code} != MQClient::MQSeries::MQRC_NONE )
	{
		($self->{err_msg}) = MQClient::MQSeries::MQReasonToText($self->{reason_code});
		return $MESSAGE_SEND_ERROR;
	}
	else
	{
		return $MESSAGE_OK;
	}
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: receiveMsg()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				receive msg from IBM MQ
#Parameters:
#	1:msgBuffer,which is the buffer will be filled with message
#	2:msgTimeOut,which is max waiting time(second).optinal,default is -1(forever)
#Return:
#				0: success
#				none 0: fail
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub receiveMsg
{
	my $self = shift;
	my ($buffer,$time_out) = @_;
	if( !defined($buffer) )
	{
		return $MESSAGE_RECEIVE_ERROR;
	}
	if( !defined($time_out) )
	{
		$time_out = -1;
	}
	my $get_msg = MQSeries::Message->new();
	my $return_value = $self->{queue}->Get
		(
			Message		=> $get_msg,
			Sync		=> 0,
			Wait		=> $time_out,
		);

	$self->{cpmplete_code} = $self->{queue}->CompCode();
	$self->{reason_code} = $self->{queue}->Reason();
	if( $return_value == -1 )
	{
		($self->{err_msg}) = MQClient::MQSeries::MQReasonToText($self->{reason_code});
		return $MESSAGE_RECEIVE_NONE;
	}
	if( $return_value == 0 )
	{
		($self->{err_msg}) = MQClient::MQSeries::MQReasonToText($self->{reason_code});
		return $MESSAGE_RECEIVE_ERROR;
	}
	$$buffer = $get_msg->Data();
	
	if( $self->{complete_code} != $MANAGER_OK || $self->{reason_code} != MQClient::MQSeries::MQRC_NONE )
	{
		($self->{err_msg}) = MQClient::MQSeries::MQReasonToText($self->{reason_code});
		if( $self->{reason_code} == MQClient::MQSeries::MQRC_NO_MSG_AVAILABLE )
		{
			return $MESSAGE_RECEIVE_NONE;
		}
		if( $self->{reason_code} == MQClient::MQSeries::MQRC_Q_MGR_NOT_AVAILABLE ||
			$self->{reason_code} == MQClient::MQSeries::MQRC_HOBJ_ERROR ||
			$self->{reason_code} == MQClient::MQSeries::MQRC_CONNECTION_BROKEN ||
			$self->{reason_code} == MQClient::MQSeries::MQRC_HCONN_ERROR ||
			$self->{reason_code} == MQClient::MQSeries::MQRC_Q_MGR_STOPPING )
		{
			return $MANAGER_NOT_OK;
		}
		if( $self->{complete_code} == MQClient::MQSeries::MQCC_FAILED )
		{
			return $MESSAGE_RECEIVE_ERROR;
		}
		return $MESSAGE_RECEIVE_INVALID;
	}
	else
	{
		$self->{msg_type} = $get_msg->MsgDesc->{MsgType};
		$self->{msg_type} -= $MSG_TYPE_FIRST;
		$self->{msg_priority} = $get_msg->MsgDesc->{Priority};
		#$self->{msg_length} = $get_msg->MsgDesc->{OriginalLength};
		$self->{msg_length} = $get_msg->BufferLength();
		return $MESSAGE_OK;
	}
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getMsgType()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current message type
#Parameters:
#				None
#Return:
#				$self->{msg_type}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getMsgType
{
	my $self = shift;
	return $self->{msg_type};
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getMsgPriority()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current message priority
#Parameters:
#				None
#Return:
#				$self->{msg_priority}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getMsgPriority
{
	my $self = shift;
	return $self->{msg_priority};
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getMsgLength()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current message length
#Parameters:
#				None
#Return:
#				$self->{msg_length}
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getMsgLength
{
	my $self = shift;
	return $self->{msg_length};
}

#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#Function Name: getAttributes()
#Author:		Chenggen Dong<[e-mail]>
#Date:			May 7 2002
#Description:
#				get current attributes
#Parameters:
#	1: msgType,current message type
#	2: msgPriority,current message priority
#	3: msgLength,current message length
#Return:
#				None
#~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
sub getAttributes
{
	my $self = shift;
	my %hash = @_;
	$hash{msgType} = $self->{msg_type};
	$hash{msgPriority} = $self->{msg_priority};
	$hash{msgLength} = $self->{msg_length};
}

1;