001: /* ----- BEGIN LICENSE BLOCK -----
002: * Version: MPL 1.1
003: *
004: * The contents of this file are subject to the Mozilla Public License Version
005: * 1.1 (the "License"); you may not use this file except in compliance with
006: * the License. You may obtain a copy of the License at
007: * http://www.mozilla.org/MPL/
008: *
009: * Software distributed under the License is distributed on an "AS IS" basis,
010: * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
011: * for the specific language governing rights and limitations under the
012: * License.
013: *
014: * The Original Code is the DataShare server.
015: *
016: * The Initial Developer of the Original Code is
017: * Ball Aerospace & Technologies Corp, Fairborn, Ohio
018: * Portions created by the Initial Developer are Copyright (C) 2001
019: * the Initial Developer. All Rights Reserved.
020: *
021: * Contributor(s): Charles Wood <cwood@ball.com>
022: *
023: * ----- END LICENSE BLOCK ----- */
024: /* RCS $Id: TcpSocket.java,v 1.3 2002/01/29 20:50:17 lizellaman Exp $
025: * $Log: TcpSocket.java,v $
026: * Revision 1.3 2002/01/29 20:50:17 lizellaman
027: * Added LoggingInterface, modified the PropertiesInterface implementation
028: *
029: * Revision 1.2 2002/01/20 23:26:29 lizellaman
030: * add command line parameter that causes an plain DataShareObject to be sent to 'inactive' TCP connections after X milliseconds on inactivity
031: *
032: * Revision 1.1.1.1 2001/10/23 13:37:18 lizellaman
033: * initial sourceforge release
034: *
035: */
036:
037: package org.datashare;
038:
039: import java.net.Socket;
040: import java.net.SocketException;
041: import java.io.EOFException;
042: import java.io.InterruptedIOException;
043: import java.io.ObjectInputStream;
044: import java.io.ObjectOutputStream;
045: import java.io.InvalidClassException;
046:
047: import org.datashare.objects.ChannelDescription;
048: import org.datashare.objects.DataShareObject;
049:
050: /**
051: * @date March 01, 2001
052: */
053: public class TcpSocket extends SocketAdapter implements Runnable {
054: private Socket socket;
055: private int priority;
056: boolean running = true;
057: private ObjectInputStream ois = null;
058: private ObjectOutputStream oos = null;
059: private String myThreadName = "";
060: private Thread myThread = null;
061: private TransmitDataThread tdt = null;
062:
063: public TcpSocket(Socket socket, DataReceiverInterface dri,
064: int priority) {
065: this .socket = socket;
066: try {
067: this .socket.setSoTimeout(SessionUtilities
068: .getTCPSocketReadTimeout()); // we may want reads to timeout
069: } catch (SocketException se) {
070: se.printStackTrace();
071: }
072: this .dri = dri;
073: this .priority = priority;
074: this .localIP = socket.getLocalAddress();
075: this .localPort = socket.getLocalPort();
076: this .remoteIP = socket.getInetAddress();
077: this .remotePort = socket.getPort();
078: this .keyValue = "Socket-TCP-" + localIP.getHostAddress() + ":"
079: + localPort + "-" + remoteIP.getHostAddress() + ":"
080: + remotePort;
081: try {
082: ois = new ObjectInputStream(socket.getInputStream());
083: oos = new ObjectOutputStream(socket.getOutputStream());
084: } catch (Exception e) {
085: e.printStackTrace();
086: stopThreadAndCloseSocket();
087: }
088:
089: SessionUtilities.getLoggingInterface().debugMsg(
090: SessionUtilities.getLoggingInterface().DEBUG,
091: SessionUtilities.getLoggingInterface().NETWORK,
092: "New connection: " + keyValue + " with active set to "
093: + (this .getActive() ? "true" : "false"));
094:
095: // create our thread for buffering and transmiting data
096: tdt = new TransmitDataThread(this );
097: tdt.setName("XmitThread for " + keyValue);
098: try {
099: tdt.setPriority(Thread.currentThread().getPriority()
100: + SessionUtilities.SocketXmtRelativePriority); // make transmitting lower priority, since it has buffering
101: } catch (Exception e) {
102: SessionUtilities.getLoggingInterface().debugMsg(
103: SessionUtilities.getLoggingInterface().ERROR,
104: SessionUtilities.getLoggingInterface().NETWORK,
105: "Problems setting Xmt Thread priority:");
106: e.printStackTrace();
107: }
108: tdt.start();
109: }
110:
111: /**
112: * returns the type socket for this instance
113: */
114: int getType() {
115: return ChannelDescription.TCP;
116: }
117:
118: /**
119: * call this method when data is to be sent over this connection, puts it into a queue
120: * so it can be sent from a xmit thread
121: */
122: void sendData(DataShareObject dsObject) {
123: SessionUtilities.getLoggingInterface().debugMsg(
124: SessionUtilities.getLoggingInterface().DEBUG,
125: SessionUtilities.getLoggingInterface().NETWORK,
126: "3-Adding " + dsObject.sendingClientKey
127: + "'s object to " + this .getClientKey()
128: + "'s xmt buffer");
129: tdt.addData(dsObject);
130: }
131:
132: /**
133: * this is the method that actually sends the data
134: */
135: protected void xmitData(DataShareObject dsObject) {
136: try {
137: if (running) {
138: oos.writeObject(dsObject);
139: oos.flush();
140: }
141: } catch (Exception e) {
142: SessionUtilities.getLoggingInterface().debugMsg(
143: SessionUtilities.getLoggingInterface().ERROR,
144: SessionUtilities.getLoggingInterface().NETWORK,
145: "Problem in Socket " + this .keyValue + ":");
146: // e.printStackTrace();
147: stopThreadAndCloseSocket();
148: }
149: }
150:
151: /**
152: * our thread loops while receiving data from Client
153: */
154: public void run() {
155: myThreadName = Thread.currentThread().getName();
156: myThread = Thread.currentThread();
157:
158: SessionUtilities.getLoggingInterface().debugMsg(
159: SessionUtilities.getLoggingInterface().DEBUG,
160: SessionUtilities.getLoggingInterface().NETWORK,
161: "Setting TCP socket " + this .keyValue + " to priority "
162: + priority);
163: try {
164: Thread.currentThread().setPriority(priority);
165: while (running) {
166: try {
167: Object object = ois.readObject(); // blocks until an object is received
168: try {
169: DataShareObject dso = (DataShareObject) object;
170: dri.clientDataReceived(dso, this ); // send it to whoever registered for these objects
171: Thread.currentThread().yield();
172: } catch (ClassCastException cce) {
173: SessionUtilities
174: .getLoggingInterface()
175: .debugMsg(
176: SessionUtilities
177: .getLoggingInterface().WARNING,
178: SessionUtilities
179: .getLoggingInterface().NETWORK,
180: "Problem-TcpSocket "
181: + this .keyValue
182: + " received unknown object");
183: }
184: } catch (ClassNotFoundException cnfe) {
185: SessionUtilities
186: .getLoggingInterface()
187: .debugMsg(
188: SessionUtilities
189: .getLoggingInterface().WARNING,
190: SessionUtilities
191: .getLoggingInterface().NETWORK,
192: "Problem reading Tcp Stream: "
193: + cnfe.getMessage());
194: } catch (InvalidClassException ice) {
195: SessionUtilities
196: .getLoggingInterface()
197: .debugMsg(
198: SessionUtilities
199: .getLoggingInterface().WARNING,
200: SessionUtilities
201: .getLoggingInterface().NETWORK,
202: "Problem reading Tcp Stream: "
203: + ice.getMessage());
204: } catch (InterruptedIOException iioe) {
205: SessionUtilities
206: .getLoggingInterface()
207: .debugMsg(
208: SessionUtilities
209: .getLoggingInterface().DEBUG,
210: SessionUtilities
211: .getLoggingInterface().NETWORK,
212: "Tcp socket " + this .keyValue
213: + " timedOut");
214: this .xmitData(new DataShareObject()); // send a packet to keep the connection alive (idle too long!)
215: }
216: }
217: SessionUtilities.getLoggingInterface().debugMsg(
218: SessionUtilities.getLoggingInterface().DEBUG,
219: SessionUtilities.getLoggingInterface().NETWORK,
220: "Thread " + Thread.currentThread().getName()
221: + " is stopping (scheduled)");
222: } catch (SocketException se) {
223: SessionUtilities.getLoggingInterface().debugMsg(
224: SessionUtilities.getLoggingInterface().DEBUG,
225: SessionUtilities.getLoggingInterface().NETWORK,
226: "Lost connection: " + keyValue);
227: stopThreadAndCloseSocket();
228: } catch (EOFException eofe) {
229: SessionUtilities.getLoggingInterface().debugMsg(
230: SessionUtilities.getLoggingInterface().DEBUG,
231: SessionUtilities.getLoggingInterface().NETWORK,
232: "Lost connection: " + keyValue);
233: stopThreadAndCloseSocket();
234: } catch (Exception e) {
235: SessionUtilities.getLoggingInterface().debugMsg(
236: SessionUtilities.getLoggingInterface().DEBUG,
237: SessionUtilities.getLoggingInterface().NETWORK,
238: "Problems reading Tcp data:");
239: e.printStackTrace();
240: stopThreadAndCloseSocket();
241: }
242: SessionUtilities.getLoggingInterface().debugMsg(
243: SessionUtilities.getLoggingInterface().DEBUG,
244: SessionUtilities.getLoggingInterface().NETWORK,
245: "Thread " + Thread.currentThread().getName()
246: + " has stopped");
247: }
248:
249: /**
250: * closes socket and stops this thread
251: */
252: void stopThreadAndCloseSocket() {
253: if (running == true) {
254: SessionUtilities.getLoggingInterface().debugMsg(
255: SessionUtilities.getLoggingInterface().DEBUG,
256: SessionUtilities.getLoggingInterface().NETWORK,
257: "removing/closing socket " + this .keyValue);
258: running = false;
259: myThread.setName("Stopped--" + myThreadName);
260: try {
261: if (socket != null)
262: socket.close();
263: } catch (Exception e) {
264: }
265: try {
266: if (oos != null)
267: oos.close();
268: } catch (Exception e) {
269: }
270: try {
271: if (ois != null)
272: ois.close();
273: } catch (Exception e) {
274: } finally {
275: tdt.stopThread();
276: dri.connectionLost(this );
277: }
278: }
279: }
280:
281: /**
282: * close socket, stop thread, and notify server
283: */
284: public void close() {
285: stopThreadAndCloseSocket();
286: }
287:
288: } //// end of class
|