udpsocketreceiver.java

一个实用工具类

/*
 * Copyright (C) butor.com. All rights reserved.
 *
 * This software is published under the terms of the GNU Library General
 * Public License (GNU LGPL), a copy of which has been included with this
 * distribution in the LICENSE.txt file. 
 */


package org.butor.net.udp.service;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.net.DatagramSocket;
import java.net.DatagramPacket;
import java.net.SocketException;

import org.butor.log.Log;
/**
 * This class listens to a port and waits
 * for a datagram. The datagram contains an operation 
 * command contained in a message string which 
 * is parsed and the processed.
 *
 * @author Filipe Mateus
 */
public class UDPSocketReceiver implements Runnable {
	public static final int BUFFER_LEN = 32;

	protected int f_udpPort = 0;
	protected int f_bufferLen = 0;
	protected IUDPSocketReceiver f_receiver = null;
	protected boolean f_shutdown = false;
	protected DatagramSocket f_socket = null;
	protected Thread f_thread = null;
	
	/**
	 * Constructs a new GatewayUDPReceiver object.
	 */
	public UDPSocketReceiver(int udpPort, int bufferLen, IUDPSocketReceiver receiver)
		throws SocketException {
		super();
		
		if (udpPort <= 0) {
			throw new SocketException("Invalid UDP port: " +udpPort);
		}
		if (bufferLen <= 0) {
			throw new SocketException("Invalid datagram length: " +bufferLen);
		}
		if (receiver == null) {
			throw new SocketException("Got null UDP receiver object!");
		}
		
		f_udpPort = udpPort;
		f_bufferLen = bufferLen;
		f_receiver = receiver;
		
		f_socket = new DatagramSocket(f_udpPort);

		start();
	}
	/**
	 * This method logs the status of the object.
	 */
	public void logDebugInfos() {
		Log.logStr(
			this,
			Log.LOG_TYPE_INFO,
			"logDebugInfos",
			"Thread is " + (f_shutdown ? "shutting down" : "running"));
			
		Log.logStr(this, Log.LOG_TYPE_INFO, "UDPReceiver", 
			"Listening for UDP message on port=" +f_udpPort
			+ ". As04Message buffer length=" +f_bufferLen +
			". Messages will be handled by " +f_receiver.getClass().getName());
	}
	/**
	 * run
	 */
	public void run() {

		if (f_socket == null) {
			return;
		}
		
		logDebugInfos();

		while (!f_shutdown) {
			byte[] buffer = new byte[f_bufferLen];
			DatagramPacket packet = new DatagramPacket(buffer, f_bufferLen);

			try {
				// set receive timeout to be able to shutdown.
				// if the jvm does not close the socket when the close() are invoked.
				f_socket.setSoTimeout(5000);
				
				f_socket.receive(packet);
				if (f_shutdown) {
					break;
				}

				String message =
					new String(packet.getData(), 0, packet.getLength());

				Log.logStr(this, Log.LOG_TYPE_INFO, "run()", 
					"Received message [" +message +"]");
				
				if ((null != message) && (message.length() > 0)) {
					f_receiver.handleUDPMessage(message);
					
				} else {
					Log.logStr(
						this,
						Log.LOG_TYPE_ERROR,
						"run",
						"Invalid UDP message [" +message +"]");
				}
				
			} catch (InterruptedIOException e) {
				// ok for timeout.
				if (f_shutdown) {
					break;
				}
				
			} catch (IOException e) {
				Log.logException(this, Log.LOG_TYPE_ERROR, "run()", e);
			}
		}

		Log.logStr(this, Log.LOG_TYPE_INFO, "run()", "Stopped.");
	}
	/**
	 * Shutdown.
	 */
	public void shutdown() {
		Log.logStr(this, Log.LOG_TYPE_INFO, "shutdown()", "Stopping ...");
		f_shutdown = true;
		
		if (f_socket != null) {
			f_socket.close();
		}
	}
	/**
	 * Start
	 */
	protected void start() {
		Log.logStr(this, Log.LOG_TYPE_INFO, "start()", "Started.");
		// give it a unique name!
		f_thread = new Thread(this, "UDPSocketReceiver_" +System.currentTimeMillis());
		f_thread.start();
	}
}