src/uk/ac/ntu/n0521366/wsyd/libs/net/NetworkStream.java

   1 /*
   2  * The MIT License
   3  *
   4  * Copyright 2015 eddie.
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a copy
   7  * of this software and associated documentation files (the "Software"), to deal
   8  * in the Software without restriction, including without limitation the rights
   9  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10  * copies of the Software, and to permit persons to whom the Software is
  11  * furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  22  * THE SOFTWARE.
  23  */
  24 package uk.ac.ntu.n0521366.wsyd.libs.net;
  25
  26 import java.io.ByteArrayOutputStream;
  27 import java.io.IOException;
  28 import java.io.InputStream;
  29 import java.io.ObjectInputStream;
  30 import java.io.ObjectOutputStream;
  31 import java.net.Socket;
  32 import java.util.List;
  33 import java.util.concurrent.ConcurrentLinkedQueue;
  34 import java.util.logging.Level;
  35 import java.util.logging.Logger;
  36 import javax.swing.SwingWorker;
  37
  38 /**
  39  *
  40  * @author eddie
  41  */
  42 public class NetworkStream extends SwingWorker<Boolean, NetworkMessage> {
  43
  44     private final Socket _socket;
  45
  46     static final long longSize = Long.SIZE / 8;
  47
  48     long _key;
  49
  50     /**
  51      * Thread safe First In, First Out Queue of NetworkMessage objects waiting to be sent.
  52      *
  53      * Allows the Owner Thread to submit new messages for sending that the Worker Thread
  54      * can safely access.
  55      */
  56     protected ConcurrentLinkedQueue<NetworkMessage> _sendMessageQueue = new ConcurrentLinkedQueue<>();
  57
  58     private final NetworkMessageEventListenerManager _eventManager;
  59
  60     private ObjectOutputStream _socketOS;
  61
  62     private InputStream _socketIS;
  63
  64     private final NetworkSocketClosing _owner;
  65
  66     private ObjectInputStream _ois;
  67
  68     private ObjectOutputStream _oos;
  69
  70     private ByteArrayOutputStream _baos;
  71
  72     public NetworkStream(Socket socket, NetworkSocketClosing owner) {
  73         this._socket = socket;
  74         this._owner = owner;
  75         _eventManager = new NetworkMessageEventListenerManager();
  76     }
  77
  78     @Override
  79     public Boolean doInBackground() throws Exception {
  80
  81         _baos = new ByteArrayOutputStream();
  82         _oos = new ObjectOutputStream(_baos);
  83         _socketOS = new ObjectOutputStream(_socket.getOutputStream());
  84
  85         _socketIS = _socket.getInputStream();
  86         _ois = new ObjectInputStream(_socketIS);
  87
  88         long expectedLength = longSize;
  89         boolean expectingLong = true;
  90
  91         while (!this.isCancelled()) {
  92             // check for incoming message
  93             if (_socketIS.available() > 0) {
  94                 System.err.println(" reading Object from stream, bytes available: " + _socketIS.available());
  95                 read();
  96             }
  97             /*
  98             if (_ois.available() >= expectedLength) {
  99                 if (expectingLong == true) {
 100                     expectedLength = _ois.readLong();
 101                     System.err.println("Expecting Object with length " + expectedLength);
 102                     System.err.println("Remaining bytes: " + _ois.available());
 103                     expectingLong = false;
 104                 }
 105                 else {
 106                     System.err.println("About to read at least " + _ois.available() + " bytes");
 107                     read();
 108                     expectedLength = longSize;
 109                     expectingLong = true;
 110                 }
 111             }
 112             */
 113             // send a queued message
 114             NetworkMessage temp = this.sendMessage();
 115             if (temp != null) {
 116                 if (!this.write(temp)) {
 117                     System.err.println("Unable to send message over TCP channel");
 118                     // FIXME: Potentially add logger support
 119                 }
 120             }
 121
 122         }
 123         return false;
 124     }
 125
 126     @Override
 127     protected void done() {
 128         _owner.canCloseSocket(_key);
 129     }
 130
 131     /**
 132      * Send a NetworkMessage over a NetworkStream.
 133      *
 134      * Each message is preceeded by its length as a long to avoid rare, but possible blocking
 135      * in the reader.
 136      *
 137      * @param message
 138      * @return true if message successfully sent
 139      */
 140     public boolean write(NetworkMessage message) {
 141         boolean result = false;
 142         System.err.println("NetworkStream.write()");
 143
 144         if (message != null) {
 145             try {
 146                 _oos.writeObject(message);
 147                 _oos.flush();
 148                 Long messageSize = new Long(_baos.size());
 149                 _socketOS.writeObject(message);
 150                 _socketOS.flush();
 151                 System.err.println(" bytes written: " + messageSize);
 152
 153                 result = true;
 154
 155             } catch (IOException ex) {
 156                 ex.printStackTrace();
 157                 // TODO: Implement graceful disconnect
 158             }
 159         }
 160         return result;
 161     }
 162
 163     /**
 164      * Reads NetworkMessage from the incoming stream and publishes it.
 165      *
 166      *
 167      * @return true if message successfully published
 168      */
 169     public boolean read() {
 170         boolean result = false;
 171         System.err.println("NetworkStream.read()");
 172
 173         try {
 174             NetworkMessage message = null;
 175             String s = null;
 176
 177             message = (NetworkMessage)_ois.readObject();
 178             System.err.println(" received: " + message.toString());
 179             message.setKey(_key);
 180             publish(message);
 181
 182             result = true;
 183         } catch (java.io.OptionalDataException ex) {
 184             System.err.println("Length: " + ex.length + " EOF: " + ex.eof);
 185             ex.printStackTrace();
 186         } catch (IOException ex) {
 187             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 188             // FIXME: Replace logger
 189             // TODO: Graceful disconnect
 190         } catch (ClassNotFoundException ex) {
 191             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 192             // FIXME: Replace logger
 193         }
 194         // TODO: Implement NetworkStream::read()
 195         return result;
 196     }
 197
 198     /**
 199      * Removes a message from the queue of pending messages.
 200      *
 201      * This method is called on the Worker Thread by the doInBackground() main loop.
 202      *
 203      * @return a message to be sent
 204      */
 205     protected NetworkMessage sendMessage() {
 206         return this._sendMessageQueue.poll();
 207     }
 208
 209     /* XXX: Methods below here all execute on the GUI Event Dispatch Thread */
 210
 211     /**
 212      * Adds a message to the queue of pending messages.
 213      *
 214      * This method will usually be called from the Owner Thread.
 215      *
 216      * @param message to be sent
 217      * @return true if the message was added to the queue
 218      */
 219     public boolean queueMessage(NetworkMessage message) throws IllegalArgumentException {
 220         boolean result = false;
 221
 222         if (message != null) {
 223             NetworkMessage temp;
 224             try { // make a deep clone of the message
 225                 temp = NetworkMessage.clone(message);
 226                 result = this._sendMessageQueue.add(temp);
 227             } catch (CloneNotSupportedException e) {
 228                 // TODO: queueMessage() log CloneNotSupportedException
 229                 e.printStackTrace();
 230             }
 231         }
 232         return result;
 233     }
 234
 235     public NetworkMessageEventListenerManager getEventManager() {
 236         return this._eventManager;
 237     }
 238
 239     /**
 240      * Fetch messages received over the stream.
 241      *
 242      * For delivery to event listeners; usually Swing GUI components. This method will run on the
 243      * Owner Thread so must complete quickly as that is the GUI Event Dispatch Thread.
 244      *
 245      * @param list messages received and queued
 246      */
 247     @Override
 248     protected void process(List<NetworkMessage> list) {
 249         for (NetworkMessage message: list) {
 250             this._eventManager.fireNetworkMessageEvent(message);
 251         }
 252     }
 253
 254 }