mqconnection.java

一个简单实用的开源C++消息中间件SAFMQ - [软件开发] - [开源 消息中间件 SAFMQ ] 2006-11-23 在很多网络应用中

/*
 Copyright 2005 Matthew J. Battey

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing, software distributed
	under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
	CONDITIONS OF ANY KIND, either express or implied. See the License for the
	specific language governing permissions and limitations under the License.


This software implements a Java interface to SAFMQ (see http://safmq.sourceforge.net).

Created on Mar 21, 2005

*/
package com.safmq;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.net.URI;
import java.util.Vector;

/**
 * This class provides an interface to communicate with a SAFMQ message queue
 * server.  Typically instances of this class are constructed by the 
 * <code>MQBuilder</code> class, however constructing the class directly
 * is also allowed.  
 * 
 * <p>Once constructed, this class may be used to construct
 * instances of the class <code>MessageQueue</code> to abstract the connection
 * interface to the message queue level.</p>
 * 
 * <p><b>Note:</b> This class is not thread safe, and applications should
 * take precautions to not initiate communications with a message queue via
 * more than one (1) thread simultaneously.</p>
 * 
 * @author Matt
 * @see MQBuilder#buildConnection(URI,String,String)
 * @see MessageQueue
 */
public class MQConnection {
	final static String FORWARD_QUEUE_NAME = "_FORWARD_";

	Socket				s;
	DataInputStream		in;
	DataOutputStream	out;
	int					server_major_protocol_version;
	int					server_minor_protocol_version;
	
	/**
	 * Result data from a call to <code>MQConnection.EnumerateQueues(Vector)</code>.
	 * Instances of this class are placed into the vector passed to 
	 * <code>EnumerateQueues()</code> to detail the specifics of the queue.
	 * 
	 * @author matt
	 * @see MQConnection#EnumerateQueues(Vector) MQConnection.EnumerateQueues(Vector)
	 */
	public class QueueData {
		String name;
		String owner;
		
		/**
		 * Constructs the QueueData object from the queue's name and owner.
		 * @param name  The name of the queue.
		 * @param owner	The name of the queue's owner.
		 */
		QueueData(String name, String owner) {
			this.name = name;
			this.owner = owner;
		}
		/**
		 * Provides the queue's name.
		 * @return The queue's name.
		 */
		public String getName() {
			return name;
		}
		/**
		 * Provides the queue's owner's name.
		 * @return	The queue's owner's name.
		 */
		public String getOwner() {
			return owner;
		}
	}
	
	/**
	 * Contains an entity's permisison when accessing a queue.
	 * 
	 * @author Matt
	 */
	public class QueuePermissions {
		String	entityName;
		boolean isgroup;
		boolean	read;
		boolean	write;
		boolean	destroy;
		boolean changeSecurity;
	
		/**
		 * Constructs the QueuePermissions from the name of the entity and 
		 * flags regarding its relationsihp with the queried queue.  This object
		 * is used as a result to 
		 * <code>MQConnection.queueEnumeratePermissions(String,Vector)</code>.
		 * 
		 * @param entityName	The name of the entity these permissions pertain to
		 * @param isgroup		A flag indicating this entity is a group (otherwise 
		 * 						a user)
		 * @param read			A flag indicating whether this entity may read from 
		 * 						the queue
		 * @param write			A flag indicating whether this entity may write to 
		 * 						the queue
		 * @param destroy		A flag indicating whether this entity may remove the 
		 * 						queue from the server
		 * @param changeSecurity	A flag indicating whether this entity may change 
		 * 							the security permissions of the queue
		 */
		QueuePermissions(String entityName, boolean isgroup, boolean read, boolean write, boolean destroy, boolean changeSecurity) {
			this.entityName = entityName;
			this.isgroup = isgroup;
			this.read = read;
			this.write = write;
			this.destroy = destroy;
			this.changeSecurity = changeSecurity;
		}
	
	
		/**
		 * Provides the name of the entity these permissions pertain to.
		 * @return Returns the name of the entity these permissions pertain to.
		 */
		public String getEntityName() {
			return entityName;
		}

		/**
		 * Provides the flag whether this entity may remove the queue from the server.
		 * 
		 * @return Returns the flag whether this entity may remove the queue from the server.
		 */
		public boolean getDestroy() {
			return destroy;
		}
		/**
		 * Provids a flag whether this entity is a group.
		 * @return Returns a flag whether this entity is a group.
		 */
		public boolean getIsGroup() {
			return isgroup;
		}
		/**
		 * Provides a flag whether this entity is allowed to read from the queue.
		 * @return Returns a flag whether this entity is allowed to read from the queue.
		 */
		public boolean getRead() {
			return read;
		}
		/**
		 * Provides a flag whether this entity is allowed to write to the queue.
		 * @return Returns a flag whether this entity is allowed to write to the queue.
		 */
		public boolean getWrite() {
			return write;
		}
		/**
		 * Provides a flag whether the entity is allowed to change the security for the queue.
		 * @return The a flag whether the entity is allowed to change the security for the queue.
		 */
		public boolean getChangeSecurity() {
			return changeSecurity;
		}
	}
	
	/**
	 * Result data from a call to <code>MQConnection.EnumerateUsers(Vector)</code>.
	 * Instances of this class are placed into the vector passed to 
	 * <code>EnumerateUsers()</code> to detail the specifics of the user.
	 * 
	 * @author matt
	 */
	public class UserDescription {
		String name;
		String description;
		
		/**
		 * Constructs the object from the user's name and description.
		 * 
		 * @param name			The name of the user
		 * @param description	The description of the user
		 */
		UserDescription(String name, String description) {
			this.name = name;
			this.description = description;
		}
		/**
		 * Provides the name of the user
		 * @return The name of the user
		 */
		public String getName() {
			return name;
		}
		/**
		 * Provides the description of the user
		 * @return The description of the user
		 */
		public String getDescription() {
			return description;
		}
		
		public boolean equals(Object o) {
			return o != null && (this==o || ((o instanceof UserDescription) && ((UserDescription)o).getName().equals(getName())) || o.equals(getName()));
		}
		
		public String toString() {
			return getClass().getName()+":["+getName()+"]";	
		}
	}
	
	/**
	 * For external use of UserDescriptions.
	 * 
	 * @param name Name of the account
	 * @param desc Description of the account
	 * @return A newly allocated <code>UserDescription</code> object.
	 */
	public UserDescription genDesc(String name, String desc) {
		return new UserDescription(name,desc);	
	}
	
	/**
	 * Constructs the object from the name of the server, the tcp/ip port to be used
	 * the user's name and password.  
	 * 
	 * @param server		The network name of the server to be contacted
	 * @param port			The TCP/IP port to be utilized in communications with the server
	 * @param user			The name of the user wishing to connect to the server
	 * @param password		The password of the user wishing to connect to the server
	 * @throws MQException	In the case that the supplied credentials are not accepted by
	 * 						the SAFMQ server. The value of <code>MQException.getErrorCode()</code>
	 * 						could be but is not limited to:
	 *<table border=0 cellpadding=3 cellspacing=0>
	 *<tr><td><code>Safmq.EC_LOGIN</code></td>
	 *				<td>The login credentials supplied were not accepted.
	 *				</td></tr>
	 *</table>
	 * 
	 * @throws IOException	In the case the SAFMQ serevr was not able to be contected
	 * 						via TCP/IP.
	 */
	public MQConnection(Socket s, String user, String password) throws MQException, IOException {
		this.s = s;
		out = new DataOutputStream(new BufferedOutputStream(s.getOutputStream(),s.getSendBufferSize()*4));
		in = new DataInputStream(new BufferedInputStream(s.getInputStream()));
			
		LOGIN_PARAMS	params = new LOGIN_PARAMS();
		paramutil.arrayCopy(params.username, user);
		paramutil.arrayCopy(params.password, password);
	
		out.write(Safmq.CMD_LOGIN);
		params.write(out);
		out.flush();
	
		int err = getResponseCode();
		if (err != Safmq.EC_NOERROR && err != Safmq.EC_NOTIMPLEMENTED) {
			throw new MQException("Login attempt failed error code (" + err+")", err);
		} else if (err == Safmq.EC_NOERROR) {
			LOGIN_RESPONSE resp = new LOGIN_RESPONSE();
			resp.read(in);
			server_major_protocol_version = resp.major_version;
			server_minor_protocol_version = resp.minor_version;
			if (server_major_protocol_version < Safmq.SAFMQ_PROTOCOL_MAJOR_VERSION)
				throw new MQException ("Login attempt failed error code ("+Safmq.EC_UNSUPPORTED_PROTOCOL+")", Safmq.EC_UNSUPPORTED_PROTOCOL);
		}
	}

	public int GetServerMajorProtocolVersion() {
		return server_major_protocol_version;
	}
		
	public int GetServerMinorProtocolVersion() {
		return server_major_protocol_version;
	}

	
	/**
	 * Opens a queue for reading and writing. The queue is then after
	 * referenced by the paramseter <code>handle</code> for calls such as 
	 * <code>Enqueue(QueueHandle,QueueMessage)</code> and 
	 * <code>Retrieve(QueueHandle,boolean,int,QueueMessage)</code>.
	 * 
	 * <p><b>Note</b>: Queues which have been opened by a call to <code>OpenQueue()</code>
	 * must be closed by a cll to <code>CloseQueue(QueueHandle)</code> if the queue
	 * is not closed, resources allocated by a call to <code>OpenQueue()</code>
	 * will not be released.</p>
	 * 
	 * @param queuename The name of the queue
	 * @param handle 	Receives a reference to the queue
	 * 
	 * @return <code>Safmq.EC_NOERROR</code> on success otherwise errors such as these, but
	 * not limited to could be produced:
	 * 
	 *<table border=0 cellpadding=3 cellspacing=0>
	 *<tr><td><code>Safmq.EC_NETWORKERROR</code></td>
	 *				<td>A networking error has occurred and the conneciton
	 *					is nolonger valid.
	 *				</td></tr>
	 *<tr><td><code>Safmq.EC_DOESNOTEXIST</code></td>
	 *				<td>The queue does not exist on the server specified.
	 *				</td></tr>
	 *</table>	 
	 *
	 *  @see #CloseQueue(QueueHandle) CloseQueue(QueueHandle handle)
	 */
	public int OpenQueue(String queuename, QueueHandle handle) {
		int ret;
		QUEUE_OPEN_QUEUE_DELETE_QUEUE_CREATE_QUEUE_ENUM_PERMS_PARAMS	params = new QUEUE_OPEN_QUEUE_DELETE_QUEUE_CREATE_QUEUE_ENUM_PERMS_PARAMS();

		paramutil.arrayCopy(params.queuename, queuename);
		try {
			out.write(Safmq.CMD_QUEUE_OPEN);
			params.write(out);
			out.flush();
			ret = getResponseCode();
			if (ret == Safmq.EC_NOERROR) {
				handle.handle = in.readInt();
			}
		} catch (IOException e) {
			ret = Safmq.EC_NETWORKERROR;
		}
		return ret;
	}
	
	/**
	 * Closes a queue which was previously opened.
	 * 
	 * @param handle Reference to the previously opened queue
	 * @return <code>Safmq.EC_NOERROR</code> on success otherwise errors such as these, but
	 * not limited to could be produced:
	 * 
	 *<table border=0 cellpadding=3 cellspacing=0>
	 *<tr><td><code>Safmq.EC_NETWORKERROR</code></td>
	 *				<td>A networking error has occurred and the conneciton
	 *					is nolonger valid.
	 *				</td></tr>
	 *<tr><td><code>Safmq.EC_NOTOPEN</code></td>
	 *				<td>The queue was not currently in an open status in the
	 *					context of this connection.
	 *				</td></tr>
	 *</table>
	 */
	public int CloseQueue(QueueHandle handle) {
		int ret = Safmq.EC_NOERROR;
		QUEUE_CLOSE_OPEN_CURSOR_PARAMS	params = new QUEUE_CLOSE_OPEN_CURSOR_PARAMS();
		params.queueID = handle.handle;