001package jmri.jmrix.srcp; 002 003import java.lang.reflect.InvocationTargetException; 004import java.util.Vector; 005import jmri.InstanceManager; 006import jmri.ShutDownManager; 007import jmri.jmrix.AbstractMRListener; 008import jmri.jmrix.AbstractMRMessage; 009import jmri.jmrix.AbstractMRReply; 010import jmri.jmrix.AbstractMRTrafficController; 011import jmri.jmrix.srcp.parser.ParseException; 012import jmri.jmrix.srcp.parser.SRCPClientParser; 013import jmri.jmrix.srcp.parser.SRCPClientVisitor; 014import jmri.jmrix.srcp.parser.SimpleNode; 015import org.slf4j.Logger; 016import org.slf4j.LoggerFactory; 017 018/** 019 * Converts Stream-based I/O to/from SRCP messages. The "SRCPInterface" side 020 * sends/receives message objects. 021 * <p> 022 * The connection to a SRCPPortController is via a pair of *Streams, which then 023 * carry sequences of characters for transmission. Note that this processing is 024 * handled in an independent thread, which we need to clean up in tests. 025 * <p> 026 * This handles the state transitions, based on the necessary state in each 027 * message. 028 * 029 * @author Bob Jacobsen Copyright (C) 2001 030 */ 031public class SRCPTrafficController extends AbstractMRTrafficController 032 implements SRCPInterface { 033 034 protected SRCPSystemConnectionMemo _memo = null; 035 final Runnable shutDownTask = () -> sendSRCPMessage(new SRCPMessage("TERM 0 SESSION"), null); 036 037 /** 038 * Create a new SRCPTrafficController instance. 039 */ 040 public SRCPTrafficController() { 041 super(); 042 InstanceManager.getDefault(ShutDownManager.class).register(shutDownTask); 043 } 044 045 // The methods to implement the SRCPInterface 046 @Override 047 public synchronized void addSRCPListener(SRCPListener l) { 048 this.addListener(l); 049 } 050 051 @Override 052 public synchronized void removeSRCPListener(SRCPListener l) { 053 this.removeListener(l); 054 } 055 056 /* 057 * Set the system connection memo associated with the traffic 058 * controller 059 */ 060 void setSystemConnectionMemo(SRCPSystemConnectionMemo memo) { 061 _memo = memo; 062 } 063 064 /* 065 * Get the system connection memo associated with the traffic 066 * controller 067 */ 068 SRCPSystemConnectionMemo getSystemConnectionMemo() { 069 return _memo; 070 } 071 072 public static int HANDSHAKEMODE = 0; 073 public static int RUNMODE = 1; 074 private int mode = HANDSHAKEMODE; 075 076 /* 077 * We are going to override the receiveLoop() function so that we can 078 * handle messages received by the system using the SRCP parser. 079 */ 080 @Override 081 public void receiveLoop() { 082 log.debug("SRCP receiveLoop starts"); 083 SRCPClientParser parser = new SRCPClientParser(istream); 084 while (true) { 085 try { 086 SimpleNode e; 087 if (_memo.getMode() == HANDSHAKEMODE) { 088 e = parser.handshakeresponse(); 089 } else { 090 e = parser.commandresponse(); 091 } 092 093 // forward the message to the registered recipients, 094 // which includes the communications monitor. 095 // return a notification via the Swing event queue to ensure proper thread 096 Runnable r = new SRCPRcvNotifier(e, mLastSender, this); 097 try { 098 javax.swing.SwingUtilities.invokeAndWait(r); 099 } catch (InterruptedException | InvocationTargetException ex) { 100 log.error("Unexpected exception in invokeAndWait:", ex); 101 } 102 log.debug("dispatch thread invoked"); 103 104 log.debug("Mode {} child contains {}", mode, ((SimpleNode) e.jjtGetChild(1)).jjtGetValue()); 105 //if (mode==HANDSHAKEMODE && ((String)((SimpleNode)e.jjtGetChild(1)).jjtGetValue()).contains("GO")) mode=RUNMODE; 106 107 SRCPClientVisitor v = new SRCPClientVisitor(); 108 e.jjtAccept(v, _memo); 109 110 // we need to re-write the switch below so that it uses the 111 // SimpleNode values instead of the reply message. 112 //SRCPReply msg = new SRCPReply((SimpleNode)e.jjtGetChild(1)); 113 switch (mCurrentState) { 114 case WAITMSGREPLYSTATE: 115 // update state, and notify to continue 116 synchronized (xmtRunnable) { 117 mCurrentState = NOTIFIEDSTATE; 118 replyInDispatch = false; 119 xmtRunnable.notify(); 120 } 121 break; 122 case WAITREPLYINPROGMODESTATE: 123 // entering programming mode 124 mCurrentMode = PROGRAMINGMODE; 125 replyInDispatch = false; 126 127 // check to see if we need to delay to allow decoders 128 // to become responsive 129 int warmUpDelay = enterProgModeDelayTime(); 130 if (warmUpDelay != 0) { 131 try { 132 synchronized (xmtRunnable) { 133 xmtRunnable.wait(warmUpDelay); 134 } 135 } catch (InterruptedException ex) { 136 Thread.currentThread().interrupt(); // retain if needed later 137 } 138 } 139 // update state, and notify to continue 140 synchronized (xmtRunnable) { 141 mCurrentState = OKSENDMSGSTATE; 142 xmtRunnable.notify(); 143 } 144 break; 145 case WAITREPLYINNORMMODESTATE: 146 // entering normal mode 147 mCurrentMode = NORMALMODE; 148 replyInDispatch = false; 149 // update state, and notify to continue 150 synchronized (xmtRunnable) { 151 mCurrentState = OKSENDMSGSTATE; 152 xmtRunnable.notify(); 153 } 154 break; 155 default: 156 replyInDispatch = false; 157 if (allowUnexpectedReply == true) { 158 log.debug("Allowed unexpected reply received in state: {} was {}", mCurrentState, e); 159 160 synchronized (xmtRunnable) { 161 // The transmit thread sometimes gets stuck 162 // when unexpected replies are received. Notify 163 // it to clear the block without a timeout. 164 // (do not change the current state) 165 xmtRunnable.notify(); 166 } 167 } else { 168 unexpectedReplyStateError(mCurrentState, e.toString()); 169 } 170 } 171 } 172 catch (ParseException pe) { 173 rcvException = true; 174 reportReceiveLoopException(pe); 175 break; 176 } 177 } 178 } 179 180 /** 181 * Terminate the SRCP extra threads. 182 * <p> 183 * This is intended to be used only by testing subclasses. 184 */ 185 @Override 186 public void terminateThreads() { 187 sendSRCPMessage(new SRCPMessage("TERM 0 SESSION"), null); 188 // we also need to remove the shutdown task. 189 jmri.InstanceManager.getDefault(jmri.ShutDownManager.class).deregister(shutDownTask); 190 // usual stop process 191 super.terminateThreads(); 192 } 193 194 /** 195 * Forward a SRCPMessage to all registered SRCPInterface listeners. 196 */ 197 @Override 198 protected void forwardMessage(AbstractMRListener client, AbstractMRMessage m) { 199 ((SRCPListener) client).message((SRCPMessage) m); 200 } 201 202 /** 203 * Forward a SRCPReply to all registered SRCPInterface listeners. 204 */ 205 @Override 206 protected void forwardReply(AbstractMRListener client, AbstractMRReply m) { 207 ((SRCPListener) client).reply((SRCPReply) m); 208 } 209 210 /** 211 * Forward a SRCPReply to all registered SRCPInterface listeners. 212 * 213 * @param client WHo should receive the reply 214 * @param n relevant node 215 */ 216 protected void forwardReply(AbstractMRListener client, SimpleNode n) { 217 ((SRCPListener) client).reply(n); 218 } 219 220 public void setSensorManager(jmri.SensorManager m) { 221 } 222 223 @Override 224 protected AbstractMRMessage pollMessage() { 225 return null; 226 } 227 228 @Override 229 protected AbstractMRListener pollReplyHandler() { 230 return null; 231 } 232 233 /** 234 * {@inheritDoc} 235 */ 236 @Override 237 public void sendSRCPMessage(SRCPMessage m, SRCPListener reply) { 238 sendMessage(m, reply); 239 } 240 241 @Override 242 protected AbstractMRMessage enterProgMode() { 243 // we need to find the right bus number! 244 return SRCPMessage.getProgMode(1); 245 } 246 247 @Override 248 protected AbstractMRMessage enterNormalMode() { 249 // we need to find the right bus number! 250 return SRCPMessage.getExitProgMode(1); 251 } 252 253 @Override 254 protected AbstractMRReply newReply() { 255 return new SRCPReply(); 256 } 257 258 @Override 259 protected boolean endOfMessage(AbstractMRReply msg) { 260 int index = msg.getNumDataElements() - 1; 261 switch (msg.getElement(index)) { 262 case 0x0D: 263 case 0x0A: 264 return true; 265 default: 266 return false; 267 } 268 } 269 270 /** 271 * Forward a "Reply" from layout to registered listeners. 272 * 273 * @param r Reply to be forwarded intact 274 * @param dest One (optional) listener to be skipped, usually because it's 275 * the originating object. 276 */ 277 @SuppressWarnings("unchecked") 278 protected void notifyReply(SimpleNode r, AbstractMRListener dest) { 279 // make a copy of the listener vector to synchronized (not needed for transmit?) 280 Vector<AbstractMRListener> v; 281 synchronized (this) { 282 // FIXME: unnecessary synchronized; the Vector IS already thread-safe. 283 v = (Vector<AbstractMRListener>) cmdListeners.clone(); 284 } 285 // forward to all listeners 286 int cnt = v.size(); 287 for (int i = 0; i < cnt; i++) { 288 AbstractMRListener client = v.elementAt(i); 289 log.debug("notify client: {}", client); 290 try { 291 //skip dest for now, we'll send the message to there last. 292 if (dest != client) { 293 forwardReply(client, r); 294 } 295 } catch (Exception ex) { 296 log.warn("notify: During reply dispatch to {}", client, ex); 297 } 298 // forward to the last listener who send a message 299 // this is done _second_ so monitoring can have already stored the reply 300 // before a response is sent 301 if (dest != null) { 302 forwardReply(dest, r); 303 } 304 } 305 } 306 307 /** 308 * Internal class to remember the Reply object and destination listener with 309 * a reply is received. 310 */ 311 protected static class SRCPRcvNotifier implements Runnable { 312 313 SimpleNode e; 314 SRCPListener mDest; 315 SRCPTrafficController mTC; 316 317 SRCPRcvNotifier(SimpleNode n, AbstractMRListener pDest, 318 AbstractMRTrafficController pTC) { 319 // the first child of n in the parse tree is 320 // the response, without the timestamp 321 e = (SimpleNode) n.jjtGetChild(1); 322 mDest = (SRCPListener) pDest; 323 mTC = (SRCPTrafficController) pTC; 324 } 325 326 @Override 327 public void run() { 328 log.debug("Delayed rcv notify starts"); 329 mTC.notifyReply(e, mDest); 330 } 331 } 332 333 private final static Logger log = LoggerFactory.getLogger(SRCPTrafficController.class); 334 335}