agateway.java

基于java开发的一套短信应用框架

	 */
	public void setFrom(String from)
	{
		this.from = from;
	}

	boolean queueMessage(OutboundMessage msg)
	{
		if (msg.getPriority() == MessagePriorities.LOW) lowQ.add(msg);
		else if (msg.getPriority() == MessagePriorities.NORMAL) normalQ.add(msg);
		else if (msg.getPriority() == MessagePriorities.HIGH) highQ.add(msg);
		return true;
	}

	int queueMessages(List msgList)
	{
		int count = 0;
		for (int i = 0, n = msgList.size(); i < n; i++)
			if (queueMessage((OutboundMessage) msgList.get(i))) count++;
		return count;
	}

	void sync() throws IOException, InterruptedException
	{
	}

	void startGateway() throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		queueManagerThread = new Thread(new QueueManager());
		queueManagerThread.start();
		started = true;
	}

	void stopGateway() throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		started = false;
		if (queueManagerThread != null)
		{
			queueManagerThread.interrupt();
			try
			{
				queueManagerThread.join();
			}
			catch (InterruptedException e)
			{
				logInfo("Interrupted while waiting for gateway to stop.");
			}
			finally
			{
				queueManagerThread = null;
			}
		}
	}

	void readMessages(List msgList, MessageClasses msgClass) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	InboundMessage readMessage(String memLoc, int memIndex) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	boolean sendMessage(OutboundMessage msg) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	int sendMessages(List msgList) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		int cnt = 0;
		for (int i = 0; i < msgList.size(); i++)
			if (sendMessage((OutboundMessage) msgList.get(i))) cnt++;
		return cnt;
	}

	boolean deleteMessage(InboundMessage msg) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	/**
	 * Queries the gateway for remaining credit.
	 * 
	 * @return Remaining credit.
	 * @throws TimeoutException
	 *             The gateway did not respond in a timely manner.
	 * @throws GatewayException
	 *             A Gateway error occurred.
	 * @throws IOException
	 *             An IO error occurred.
	 * @throws InterruptedException
	 *             The call was interrupted.
	 */
	public float queryBalance() throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	/**
	 * Queries the gateway to see if a specific message and its recipient are
	 * covered. The given message is not sent out - it is just tested.
	 * 
	 * @param msg
	 *            The message to test.
	 * @return True is the recipient is covered by the network.
	 * @throws TimeoutException
	 *             The gateway did not respond in a timely manner.
	 * @throws GatewayException
	 *             A Gateway error occurred.
	 * @throws IOException
	 *             An IO error occurred.
	 * @throws InterruptedException
	 *             The call was interrupted.
	 */
	public boolean queryCoverage(OutboundMessage msg) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	/**
	 * Query the gateway for message delivery status.
	 * 
	 * @param msg
	 *            The OutboundMessage object to be checked.
	 * @return The delivery status. This is interpreted and mapped to the
	 *         standard SMSLib status codes. For detailed information, check
	 *         method getDeliveryErrorCode().
	 * @throws TimeoutException
	 *             The gateway did not respond in a timely manner.
	 * @throws GatewayException
	 *             A Gateway error occurred.
	 * @throws IOException
	 *             An IO error occurred.
	 * @throws InterruptedException
	 *             The call was interrupted.
	 * @see DeliveryStatuses
	 * @see #getDeliveryErrorCode()
	 */
	public DeliveryStatuses queryMessage(OutboundMessage msg) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		return queryMessage(msg.getRefNo());
	}

	/**
	 * Query the gateway for message delivery status.
	 * 
	 * @param refNo
	 *            The reference number of a previously sent message to be
	 *            checked.
	 * @return The delivery status. This is interpreted and mapped to the
	 *         standard SMSLib status codes. For detailed information, check
	 *         method getDeliveryErrorCode().
	 * @throws TimeoutException
	 *             The gateway did not respond in a timely manner.
	 * @throws GatewayException
	 *             A Gateway error occurred.
	 * @throws IOException
	 *             An IO error occurred.
	 * @throws InterruptedException
	 *             The call was interrupted.
	 * @see DeliveryStatuses
	 * @see #getDeliveryErrorCode()
	 */
	public DeliveryStatuses queryMessage(String refNo) throws TimeoutException, GatewayException, IOException, InterruptedException
	{
		throw new GatewayException("Feature not supported.");
	}

	/**
	 * Returns the gateway-specific error code from the last queryMessage()
	 * call. Note that each call to queryMessage() resets this error.
	 * 
	 * @return The error code - actual values depend on gateway used.
	 * @see #queryMessage(OutboundMessage)
	 */
	public int getDeliveryErrorCode()
	{
		return deliveryErrorCode;
	}

	boolean isCapableOf(int att)
	{
		return ((att & attributes) == att);
	}

	boolean conformsTo(int attrib, boolean required)
	{
		if (required && !isCapableOf(attrib)) return false;
		else return true;
	}

	static class Statistics
	{
		public int inbound;

		public int outbound;

		public Statistics()
		{
			inbound = 0;
			outbound = 0;
		}
	}

	int getQueueLoad()
	{
		return (getQueueLoad(MessagePriorities.LOW) + getQueueLoad(MessagePriorities.NORMAL) + getQueueLoad(MessagePriorities.HIGH));
	}

	int getQueueLoad(MessagePriorities priority)
	{
		if (priority == MessagePriorities.LOW) return lowQ.size();
		else if (priority == MessagePriorities.NORMAL) return normalQ.size();
		else if (priority == MessagePriorities.HIGH) return highQ.size();
		else return 0;
	}

	int getQueueDelay()
	{
		return queueDelay;
	}

	void setQueueDelay(int queueDelay)
	{
		this.queueDelay = queueDelay;
	}

	int getQueueRetries()
	{
		return queueRetries;
	}

	void setQueueRetries(int queueRetries)
	{
		this.queueRetries = queueRetries;
	}

	String formatLog(String message)
	{
		return "Gateway: " + gtwId + " " + message;
	}

	void logError(String message)
	{
		logger.error(formatLog(message));
	}

	void logDebug(String message)
	{
		logger.debug(formatLog(message));
	}

	void logInfo(String message)
	{
		logger.info(formatLog(message));
	}

	private class QueueManager implements Runnable
	{
		public QueueManager()
		{
			super();
		}

		public Object get()
		{
			if (highQ.size() > 0) return highQ.get();
			else if (normalQ.size() > 0) return normalQ.get();
			else if (lowQ.size() > 0) return lowQ.get();
			else return null;
		}

		public void run()
		{
			OutboundMessage msg;
			logInfo("Starting Queue Manager");
			try
			{
				if (started)
				{
					while (true)
					{
						while (true)
						{
							msg = (OutboundMessage) get();
							if (msg == null) Thread.sleep(queueDelay);
							else break;
						}
						if (!sendMessage(msg))
						{
							if (msg != null)
							{
								if (msg.getRetryCount() < queueRetries)
								{
									logInfo("Reinserting message to queue.");
									msg.incrementRetryCount();
									queueMessage(msg);
								}
								else
								{
									if (getOutboundNotification() != null) getOutboundNotification().process(gtwId, msg);
									logInfo("Maximum number of retries (" + queueRetries + ") exceeded.");
									msg.setFailureCause(FailureCauses.UNKNOWN);
								}
							}
						}
						else if (getOutboundNotification() != null) getOutboundNotification().process(gtwId, msg);
						Thread.sleep(queueDelay);
					}
				}
			}
			catch (InterruptedException e)
			{
				logInfo("Interrupting queue.");
			}
			catch (Exception e)
			{
				logInfo("Interrupting queue.");
			}
			logInfo("QueueManager stopped");
		}
	}
}