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
  93             // check for incoming message
  94             if (_ois.available() >= expectedLength) {
  95                 if (expectingLong == true) {
  96                     expectedLength = _ois.readLong();
  97                     System.err.println("Expecting Object with length " + expectedLength);
  98                     System.err.println("Remaining bytes: " + _ois.available());
  99                     expectingLong = false;
 100                 }
 101                 else {
 102                     System.err.println("About to read at least " + _ois.available() + " bytes");
 103                     read();
 104                     expectedLength = longSize;
 105                     expectingLong = true;
 106                 }
 107             }
 108             // send a queued message
 109             NetworkMessage temp = this.sendMessage();
 110             if (temp != null) {
 111                 if (!this.write(temp)) {
 112                     System.err.println("Unable to send message over TCP channel");
 113                     // FIXME: Potentially add logger support
 114                 }
 115             }
 116
 117         }
 118         return false;
 119     }
 120
 121     @Override
 122     protected void done() {
 123         _owner.canCloseSocket(_key);
 124     }
 125
 126     /**
 127      * Send a NetworkMessage over a NetworkStream.
 128      *
 129      * Each message is preceeded by its length as a long to avoid rare, but possible blocking
 130      * in the reader.
 131      *
 132      * @param message
 133      * @return true if message successfully sent
 134      */
 135     public boolean write(NetworkMessage message) {
 136         boolean result = false;
 137         System.err.println("NetworkStream.write()");
 138
 139         if (message != null) {
 140             try {
 141                 _oos.writeObject(message);
 142                 _oos.flush();
 143                 long messageSize = _baos.size();
 144                 _socketOS.writeLong(messageSize);
 145                 System.err.println(" bytes to write: " + messageSize);
 146                 _baos.writeTo(_socketOS);
 147                 _socketOS.flush();
 148                 System.err.println("baos.size()=" + _baos.size());
 149
 150                 result = true;
 151
 152             } catch (IOException ex) {
 153                 ex.printStackTrace();
 154                 // TODO: Implement graceful disconnect
 155             }
 156         }
 157         return result;
 158     }
 159
 160     /**
 161      * Reads NetworkMessage from the incoming stream and publishes it.
 162      *
 163      *
 164      * @return true if message successfully published
 165      */
 166     public boolean read() {
 167         boolean result = false;
 168         System.err.println("NetworkStream.read()");
 169
 170         try {
 171             NetworkMessage message = null;
 172
 173             try {
 174                 message = (NetworkMessage)_ois.readObject();
 175             } catch (java.io.OptionalDataException ex) {
 176                 System.err.println("Length: " + ex.length + " EOF: " + ex.eof);
 177                 ex.printStackTrace();
 178             }
 179             message.setKey(_key);
 180             publish(message);
 181
 182             result = true;
 183         } catch (IOException ex) {
 184             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 185             // FIXME: Replace logger
 186             // TODO: Graceful disconnect
 187         } catch (ClassNotFoundException ex) {
 188             Logger.getLogger(NetworkStream.class.getName()).log(Level.SEVERE, null, ex);
 189             // FIXME: Replace logger
 190         }
 191         // TODO: Implement NetworkStream::read()
 192         return result;
 193     }
 194
 195     /**
 196      * Removes a message from the queue of pending messages.
 197      *
 198      * This method is called on the Worker Thread by the doInBackground() main loop.
 199      *
 200      * @return a message to be sent
 201      */
 202     protected NetworkMessage sendMessage() {
 203         return this._sendMessageQueue.poll();
 204     }
 205
 206     /* XXX: Methods below here all execute on the GUI Event Dispatch Thread */
 207
 208     /**
 209      * Adds a message to the queue of pending messages.
 210      *
 211      * This method will usually be called from the Owner Thread.
 212      *
 213      * @param message to be sent
 214      * @return true if the message was added to the queue
 215      */
 216     public boolean queueMessage(NetworkMessage message) throws IllegalArgumentException {
 217         boolean result = false;
 218
 219         if (message != null) {
 220             NetworkMessage temp;
 221             try { // make a deep clone of the message
 222                 temp = NetworkMessage.clone(message);
 223                 result = this._sendMessageQueue.add(temp);
 224             } catch (CloneNotSupportedException e) {
 225                 // TODO: queueMessage() log CloneNotSupportedException
 226                 e.printStackTrace();
 227             }
 228         }
 229         return result;
 230     }
 231
 232     public NetworkMessageEventListenerManager getEventManager() {
 233         return this._eventManager;
 234     }
 235
 236     /**
 237      * Fetch messages received over the stream.
 238      *
 239      * For delivery to event listeners; usually Swing GUI components. This method will run on the
 240      * Owner Thread so must complete quickly as that is the GUI Event Dispatch Thread.
 241      *
 242      * @param list messages received and queued
 243      */
 244     @Override
 245     protected void process(List<NetworkMessage> list) {
 246         for (NetworkMessage message: list) {
 247             this._eventManager.fireNetworkMessageEvent(message);
 248         }
 249     }
 250
 251 }