001package jmri.jmrix.roco.z21; 002 003import java.io.DataInputStream; 004import java.io.DataOutputStream; 005import java.io.IOException; 006import java.io.PipedInputStream; 007import java.io.PipedOutputStream; 008import jmri.jmrix.lenz.XNetListener; 009import jmri.jmrix.lenz.XNetMessage; 010import jmri.jmrix.lenz.XNetReply; 011import jmri.util.ImmediatePipedOutputStream; 012import org.slf4j.Logger; 013import org.slf4j.LoggerFactory; 014 015/** 016 * Interface between z21 messages and an XpressNet stream. 017 * <p> 018 * Parts of this code are derived from the 019 * jmri.jmrix.lenz.xnetsimulator.XNetSimulatorAdapter class. 020 * 021 * @author Paul Bender Copyright (C) 2014 022 */ 023public class Z21XPressNetTunnel implements Z21Listener, XNetListener, Runnable { 024 025 jmri.jmrix.lenz.XNetStreamPortController xsc = null; 026 private DataOutputStream pout = null; // for output to other classes 027 private DataInputStream pin = null; // for input from other classes 028 // internal ends of the pipes 029 private DataOutputStream outpipe = null; // feed pin 030 private DataInputStream inpipe = null; // feed pout 031 private Z21SystemConnectionMemo _memo; 032 private Thread sourceThread; 033 034 /** 035 * Build a new XpressNet tunnel. 036 * @param memo system connection. 037 */ 038 @edu.umd.cs.findbugs.annotations.SuppressFBWarnings(value="SC_START_IN_CTOR", justification="done at end, waits for data") 039 public Z21XPressNetTunnel(Z21SystemConnectionMemo memo) { 040 // save the SystemConnectionMemo. 041 _memo = memo; 042 043 // configure input and output pipes to use for 044 // the communication with the XpressNet implementation. 045 try { 046 PipedOutputStream tempPipeI = new ImmediatePipedOutputStream(); 047 pout = new DataOutputStream(tempPipeI); 048 inpipe = new DataInputStream(new PipedInputStream(tempPipeI)); 049 PipedOutputStream tempPipeO = new ImmediatePipedOutputStream(); 050 outpipe = new DataOutputStream(tempPipeO); 051 pin = new DataInputStream(new PipedInputStream(tempPipeO)); 052 } catch (java.io.IOException e) { 053 log.error("init (pipe): Exception: {}", e.toString()); 054 return; 055 } 056 057 // start a thread to read from the input pipe. 058 sourceThread = new Thread(this); 059 sourceThread.setName("z21.Z21XpressNetTunnel sourceThread"); 060 sourceThread.setDaemon(true); 061 sourceThread.start(); 062 063 // Then use those pipes as the input and output pipes for 064 // a new XNetStreamPortController object. 065 setStreamPortController(new Z21XNetStreamPortController(pin, pout, "None")); 066 067 // register as a Z21Listener, so we can receive replies 068 _memo.getTrafficController().addz21Listener(this); 069 070 // start the XpressNet configuration. 071 xsc.configure(); 072 } 073 074 @Override 075 public void run() { // start a new thread 076 // this thread has one task. It repeatedly reads from the input pipe 077 // and writes modified data to the output pipe. This is the heart 078 // of the command station simulation. 079 log.debug("Simulator Thread Started"); 080 for (;;) { 081 Z21XNetMessage m = readMessage(); 082 if(m != null) { 083 // don't forward a null message. 084 message(m); 085 } 086 } 087 } 088 089 /** 090 * Read one incoming message from the buffer and set 091 * outputBufferEmpty to true. 092 */ 093 private Z21XNetMessage readMessage() { 094 Z21XNetMessage msg = null; 095 try { 096 msg = loadChars(); 097 } catch (java.io.IOException e) { 098 // should do something meaningful here. 099 100 } 101 return (msg); 102 } 103 104 /** 105 * Get characters from the input source, and file a message. 106 * <p> 107 * Returns only when the message is complete. 108 * <p> 109 * Only used in the Receive thread. 110 * 111 * @return filled message 112 * @throws IOException when presented by the input source. 113 */ 114 private Z21XNetMessage loadChars() throws java.io.IOException { 115 int i; 116 byte char1; 117 char1 = readByteProtected(inpipe); 118 int len = (char1 & 0x0f) + 2; // opCode+Nbytes+ECC 119 // The z21 protocol has a special case for 120 // LAN_X_GET_TURNOUT_INFO, which advertises as having 121 // 3 payload bytes, but really only has two. 122 if((char1&0xff)==Z21Constants.LAN_X_GET_TURNOUT_INFO) 123 { 124 len=4; 125 } 126 Z21XNetMessage msg = new Z21XNetMessage(len); 127 msg.setElement(0, char1 & 0xFF); 128 for (i = 1; i < len; i++) { 129 char1 = readByteProtected(inpipe); 130 msg.setElement(i, char1 & 0xFF); 131 } 132 return msg; 133 } 134 135 /** 136 * Read a single byte, protecting against various timeouts, etc. 137 * <p> 138 * When a port is set to have a receive timeout (via the 139 * enableReceiveTimeout() method), some will return zero bytes or an 140 * EOFException at the end of the timeout. In that case, the read should be 141 * repeated to get the next real character. 142 */ 143 private byte readByteProtected(DataInputStream istream) throws java.io.IOException { 144 byte[] rcvBuffer = new byte[1]; 145 while (true) { // loop will repeat until character found 146 int nchars; 147 nchars = istream.read(rcvBuffer, 0, 1); 148 if (nchars > 0) { 149 return rcvBuffer[0]; 150 } 151 } 152 } 153 154 // Z21Listener interface methods. 155 156 /** 157 * Member function that will be invoked by a z21Interface implementation to 158 * forward a z21 message from the layout. 159 * 160 * @param msg The received z21 message. Note that this same object may be 161 * presented to multiple users. It should not be modified here. 162 */ 163 @Override 164 public void reply(Z21Reply msg) { 165 // This funcction forwards the payload of an XpressNet message 166 // tunneled in a z21 message and forwards it to the XpressNet 167 // implementation's input stream. 168 if (msg.isXPressNetTunnelMessage()) { 169 Z21XNetReply reply = msg.getXNetReply(); 170 log.debug("Z21 Reply {} forwarded to XpressNet implementation as {}", 171 msg, reply); 172 for (int i = 0; i < reply.getNumDataElements(); i++) { 173 try { 174 outpipe.writeByte(reply.getElement(i)); 175 } catch (java.io.IOException ioe) { 176 log.error("Error writing XpressNet Reply to XpressNet input stream."); 177 } 178 } 179 } 180 } 181 182 /** 183 * Member function that will be invoked by a z21Interface implementation to 184 * forward a z21 message sent to the layout. Normally, this function will do 185 * nothing. 186 * 187 * @param msg The received z21 message. Note that this same object may be 188 * presented to multiple users. It should not be modified here. 189 */ 190 @Override 191 public void message(Z21Message msg) { 192 // this function does nothing. 193 } 194 195 // XNetListener Interface methods. 196 197 /** 198 * Member function that will be invoked by a XNetInterface implementation to 199 * forward a XNet message from the layout. 200 * 201 * @param msg The received XNet message. Note that this same object may be 202 * presented to multiple users. It should not be modified here. 203 */ 204 @Override 205 public void message(XNetReply msg) { 206 // we don't do anything with replies. 207 } 208 209 /** 210 * Member function that will be invoked by a XNetInterface implementation to 211 * forward a XNet message sent to the layout. Normally, this function will 212 * do nothing. 213 * 214 * @param msg The received XNet message. Note that this same object may be 215 * presented to multiple users. It should not be modified here. 216 */ 217 @Override 218 public void message(XNetMessage msg) { 219 // when an XpressNet message shows up here, package it in a Z21Message 220 Z21Message message = new Z21Message(msg); 221 log.debug("XpressNet Message {} forwarded to z21 Interface as {}", 222 msg, message); 223 // and send the z21 message to the interface 224 _memo.getTrafficController().sendz21Message(message, this); 225 } 226 227 /** 228 * Member function invoked by an XNetInterface implementation to notify a 229 * sender that an outgoing message timed out and was dropped from the queue. 230 */ 231 @Override 232 public void notifyTimeout(XNetMessage msg) { 233 // we don't do anything with timeouts. 234 } 235 236 /** 237 * Package protected method to retrieve the stream port controller 238 * associated with this tunnel. 239 * @return controller in use 240 */ 241 jmri.jmrix.lenz.XNetStreamPortController getStreamPortController() { 242 return xsc; 243 } 244 245 /** 246 * Package protected method to set the stream port controller 247 * associated with this tunnel. 248 * @param x controller to retain 249 */ 250 void setStreamPortController(jmri.jmrix.lenz.XNetStreamPortController x){ 251 xsc = x; 252 253 // configure the XpressNet connections properties. 254 xsc.getSystemConnectionMemo().setSystemPrefix("X"); 255 xsc.getSystemConnectionMemo().setUserName(_memo.getUserName() + "XpressNet"); 256 257 // register a connection config object for this stream port. 258 //jmri.InstanceManager.getDefault(jmri.ConfigureManager.class).registerPref(new Z21XNetConnectionConfig(xsc)); 259 //jmri.InstanceManager.getDefault(jmri.jmrix.ConnectionConfigManager.class).add(new Z21XNetConnectionConfig(xsc)); 260 } 261 262 @SuppressWarnings("deprecation") // Thread.stop 263 public void dispose(){ 264 if(xsc != null){ 265 xsc.dispose(); 266 } 267 try { 268 inpipe.close(); 269 } catch (IOException ex) { 270 // Ignore IO error 271 } 272 } 273 274 private final static Logger log = LoggerFactory.getLogger(Z21XPressNetTunnel.class); 275 276}