Moving a recently added trim() call in SwitchHandler after Null check.
[controller.git] / opendaylight / protocol_plugins / openflow / src / main / java / org / opendaylight / controller / protocol_plugin / openflow / core / internal / SwitchHandler.java
1
2 /*
3  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
4  *
5  * This program and the accompanying materials are made available under the
6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
7  * and is available at http://www.eclipse.org/legal/epl-v10.html
8  */
9
10 package org.opendaylight.controller.protocol_plugin.openflow.core.internal;
11
12 import java.io.IOException;
13 import java.nio.channels.AsynchronousCloseException;
14 import java.nio.channels.SelectionKey;
15 import java.nio.channels.Selector;
16 import java.nio.channels.SocketChannel;
17 import java.nio.channels.spi.SelectorProvider;
18 import java.util.ArrayList;
19 import java.util.Comparator;
20 import java.util.Date;
21 import java.util.HashMap;
22 import java.util.Iterator;
23 import java.util.List;
24 import java.util.Map;
25 import java.util.Set;
26 import java.util.Timer;
27 import java.util.TimerTask;
28 import java.util.concurrent.Callable;
29 import java.util.concurrent.ConcurrentHashMap;
30 import java.util.concurrent.ExecutorService;
31 import java.util.concurrent.Executors;
32 import java.util.concurrent.Future;
33 import java.util.concurrent.PriorityBlockingQueue;
34 import java.util.concurrent.TimeUnit;
35 import java.util.concurrent.atomic.AtomicInteger;
36
37 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
38 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
39 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageReadWrite;
40 import org.openflow.protocol.OFBarrierReply;
41 import org.openflow.protocol.OFEchoReply;
42 import org.openflow.protocol.OFError;
43 import org.openflow.protocol.OFFeaturesReply;
44 import org.openflow.protocol.OFFlowMod;
45 import org.openflow.protocol.OFGetConfigReply;
46 import org.openflow.protocol.OFMatch;
47 import org.openflow.protocol.OFMessage;
48 import org.openflow.protocol.OFPhysicalPort;
49 import org.openflow.protocol.OFPhysicalPort.OFPortConfig;
50 import org.openflow.protocol.OFPhysicalPort.OFPortFeatures;
51 import org.openflow.protocol.OFPhysicalPort.OFPortState;
52 import org.openflow.protocol.OFPort;
53 import org.openflow.protocol.OFPortStatus;
54 import org.openflow.protocol.OFPortStatus.OFPortReason;
55 import org.openflow.protocol.OFSetConfig;
56 import org.openflow.protocol.OFStatisticsReply;
57 import org.openflow.protocol.OFStatisticsRequest;
58 import org.openflow.protocol.OFType;
59 import org.openflow.protocol.factory.BasicFactory;
60 import org.openflow.util.HexString;
61 import org.slf4j.Logger;
62 import org.slf4j.LoggerFactory;
63
64 public class SwitchHandler implements ISwitch {
65     private static final Logger logger = LoggerFactory
66             .getLogger(SwitchHandler.class);
67     private static final int SWITCH_LIVENESS_TIMER = 5000;
68     private static final int SWITCH_LIVENESS_TIMEOUT = 2 * SWITCH_LIVENESS_TIMER + 500;
69     private int MESSAGE_RESPONSE_TIMER = 2000;
70
71     private String instanceName;
72     private ISwitch thisISwitch;
73     private IController core;
74     private Long sid;
75     private Integer buffers;
76     private Integer capabilities;
77     private Byte tables;
78     private Integer actions;
79     private Selector selector;
80     private SocketChannel socket;
81     private BasicFactory factory;
82     private AtomicInteger xid;
83     private SwitchState state;
84     private Timer periodicTimer;
85     private Map<Short, OFPhysicalPort> physicalPorts;
86     private Map<Short, Integer> portBandwidth;
87     private Date connectedDate;
88     private Long lastMsgReceivedTimeStamp;
89     private Boolean probeSent;
90     private ExecutorService executor;
91     private ConcurrentHashMap<Integer, Callable<Object>> messageWaitingDone;
92     private boolean running;
93     private IMessageReadWrite msgReadWriteService;
94     private Thread switchHandlerThread;
95     private Integer responseTimerValue;
96         private PriorityBlockingQueue<PriorityMessage> transmitQ;
97     private Thread transmitThread;
98     
99     private enum SwitchState {
100         NON_OPERATIONAL(0), WAIT_FEATURES_REPLY(1), WAIT_CONFIG_REPLY(2), OPERATIONAL(
101                 3);
102
103         private int value;
104
105         private SwitchState(int value) {
106             this.value = value;
107         }
108
109         @SuppressWarnings("unused")
110         public int value() {
111             return this.value;
112         }
113     }
114
115     public SwitchHandler(Controller core, SocketChannel sc, String name) {
116         this.instanceName = name;
117         this.thisISwitch = this;
118         this.sid = (long) 0;
119         this.buffers = (int)0;
120         this.capabilities = (int)0;
121         this.tables = (byte)0;
122         this.actions = (int)0;
123         this.core = core;
124         this.socket = sc;
125         this.factory = new BasicFactory();
126         this.connectedDate = new Date();
127         this.lastMsgReceivedTimeStamp = connectedDate.getTime();
128         this.physicalPorts = new HashMap<Short, OFPhysicalPort>();
129         this.portBandwidth = new HashMap<Short, Integer>();
130         this.state = SwitchState.NON_OPERATIONAL;
131         this.probeSent = false;
132         this.xid = new AtomicInteger(this.socket.hashCode());
133         this.periodicTimer = null;
134         this.executor = Executors.newFixedThreadPool(4);
135         this.messageWaitingDone = new ConcurrentHashMap<Integer, Callable<Object>>();
136         this.responseTimerValue = MESSAGE_RESPONSE_TIMER;
137         String rTimer = System.getProperty("of.messageResponseTimer");
138         if (rTimer != null) {
139                 try {
140                         responseTimerValue = Integer.decode(rTimer);
141                 } catch (NumberFormatException e) {
142                                 logger.warn("Invalid of.messageResponseTimer: {} use default({})",
143                                                 rTimer, MESSAGE_RESPONSE_TIMER);
144                 }
145         }
146         }
147
148     public void start() {
149         try {
150                 startTransmitThread();
151                 setupCommChannel();
152                 sendFirstHello();
153             startHandlerThread();
154         } catch (Exception e) {
155             reportError(e);
156         }
157     }
158
159     private void startHandlerThread() {
160         switchHandlerThread = new Thread(new Runnable() {
161             @Override
162             public void run() {
163                 running = true;
164                 while (running) {
165                     try {
166                         // wait for an incoming connection
167                         selector.select(0);
168                         Iterator<SelectionKey> selectedKeys = selector
169                                 .selectedKeys().iterator();
170                         while (selectedKeys.hasNext()) {
171                             SelectionKey skey = selectedKeys.next();
172                             selectedKeys.remove();
173                             if (skey.isValid() && skey.isWritable()) {
174                                 resumeSend();
175                             }
176                             if (skey.isValid() && skey.isReadable()) {
177                                 handleMessages();
178                             }
179                         }
180                     } catch (Exception e) {
181                         reportError(e);
182                     }
183                 }
184             }
185         }, instanceName);
186         switchHandlerThread.start();
187     }
188
189     public void stop() {
190         running = false;
191         cancelSwitchTimer();
192         try {
193                 selector.wakeup();
194                 selector.close();
195                 } catch (Exception e) {
196                 }
197         try {
198                         socket.close();
199                 } catch (Exception e) {
200                 }
201         try {
202                         msgReadWriteService.stop();
203                 } catch (Exception e) {
204                 }
205         executor.shutdown();
206         
207         selector = null;
208         socket = null;
209                 msgReadWriteService = null;
210                 
211                 if (switchHandlerThread != null) {
212                         switchHandlerThread.interrupt();
213                 }
214                 if (transmitThread != null) {
215                         transmitThread.interrupt();
216                 }
217     }
218
219     @Override
220     public int getNextXid() {
221         return this.xid.incrementAndGet();
222     }
223
224         /**
225          * This method puts the message in an outgoing priority queue with normal
226          * priority. It will be served after high priority messages. The method
227          * should be used for non-critical messages such as statistics request,
228          * discovery packets, etc. An unique XID is generated automatically and
229          * inserted into the message.
230          * 
231          * @param msg The OF message to be sent
232          * @return The XID used
233          */
234     @Override
235     public Integer asyncSend(OFMessage msg) {
236         return asyncSend(msg, getNextXid());
237     }
238
239         /**
240          * This method puts the message in an outgoing priority queue with normal
241          * priority. It will be served after high priority messages. The method
242          * should be used for non-critical messages such as statistics request,
243          * discovery packets, etc. The specified XID is inserted into the message.
244          * 
245          * @param msg The OF message to be Sent
246          * @param xid The XID to be used in the message
247          * @return The XID used
248          */
249     @Override
250     public Integer asyncSend(OFMessage msg, int xid) {
251         msg.setXid(xid);
252         transmitQ.add(new PriorityMessage(msg, 0));
253         return xid;
254     }
255
256         /**
257          * This method puts the message in an outgoing priority queue with high
258          * priority. It will be served first before normal priority messages. The
259          * method should be used for critical messages such as hello, echo reply
260          * etc. An unique XID is generated automatically and inserted into the
261          * message.
262          * 
263          * @param msg The OF message to be sent
264          * @return The XID used
265          */
266     @Override
267     public Integer asyncFastSend(OFMessage msg) {
268         return asyncFastSend(msg, getNextXid());
269     }
270
271         /**
272          * This method puts the message in an outgoing priority queue with high
273          * priority. It will be served first before normal priority messages. The
274          * method should be used for critical messages such as hello, echo reply
275          * etc. The specified XID is inserted into the message.
276          * 
277          * @param msg The OF message to be sent
278          * @return The XID used
279          */
280     @Override
281     public Integer asyncFastSend(OFMessage msg, int xid) {
282         msg.setXid(xid);
283         transmitQ.add(new PriorityMessage(msg, 1));
284         return xid;
285     }
286
287    public void resumeSend() {
288         try {
289                         msgReadWriteService.resumeSend();
290                 } catch (Exception e) {
291                         reportError(e);
292                 }
293     }
294
295     public void handleMessages() {
296         List<OFMessage> msgs = null;
297         
298         try {
299                 msgs = msgReadWriteService.readMessages();
300                 } catch (Exception e) {
301                         reportError(e);
302                 }
303                 
304         if (msgs == null) {
305             logger.debug("{} is down", toString());
306             // the connection is down, inform core
307             reportSwitchStateChange(false);
308             return;
309         }
310         for (OFMessage msg : msgs) {
311             logger.trace("Message received: {}", msg.toString());
312             /*
313             if  ((msg.getType() != OFType.ECHO_REQUEST) &&
314                         (msg.getType() != OFType.ECHO_REPLY)) {
315                 logger.debug(msg.getType().toString() + " received from sw " + toString());
316             }
317              */
318             this.lastMsgReceivedTimeStamp = System.currentTimeMillis();
319             OFType type = msg.getType();
320             switch (type) {
321             case HELLO:
322                 // send feature request
323                 OFMessage featureRequest = factory
324                         .getMessage(OFType.FEATURES_REQUEST);
325                 asyncFastSend(featureRequest);
326                 // delete all pre-existing flows
327                 OFMatch match = new OFMatch().setWildcards(OFMatch.OFPFW_ALL);
328                 OFFlowMod flowMod = (OFFlowMod) factory
329                         .getMessage(OFType.FLOW_MOD);
330                 flowMod.setMatch(match).setCommand(OFFlowMod.OFPFC_DELETE)
331                         .setOutPort(OFPort.OFPP_NONE).setLength(
332                                 (short) OFFlowMod.MINIMUM_LENGTH);
333                 asyncFastSend(flowMod);
334                 this.state = SwitchState.WAIT_FEATURES_REPLY;
335                 startSwitchTimer();
336                 break;
337             case ECHO_REQUEST:
338                 OFEchoReply echoReply = (OFEchoReply) factory
339                         .getMessage(OFType.ECHO_REPLY);
340                 asyncFastSend(echoReply);
341                 break;
342             case ECHO_REPLY:
343                 this.probeSent = false;
344                 break;
345             case FEATURES_REPLY:
346                 processFeaturesReply((OFFeaturesReply) msg);
347                 break;
348             case GET_CONFIG_REPLY:
349                 // make sure that the switch can send the whole packet to the controller
350                 if (((OFGetConfigReply) msg).getMissSendLength() == (short) 0xffff) {
351                     this.state = SwitchState.OPERATIONAL;
352                 }
353                 break;
354             case BARRIER_REPLY:
355                 processBarrierReply((OFBarrierReply) msg);
356                 break;
357             case ERROR:
358                 processErrorReply((OFError) msg);
359                 break;
360             case PORT_STATUS:
361                 processPortStatusMsg((OFPortStatus) msg);
362                 break;
363             case STATS_REPLY:
364                 processStatsReply((OFStatisticsReply) msg);
365                 break;
366             case PACKET_IN:
367                 break;
368             default:
369                 break;
370             } // end of switch
371             if (isOperational()) {
372                 ((Controller) core).takeSwitchEventMsg(thisISwitch, msg);
373             }
374         } // end of for
375     }
376
377     private void processPortStatusMsg(OFPortStatus msg) {
378         //short portNumber = msg.getDesc().getPortNumber();
379         OFPhysicalPort port = msg.getDesc();
380         if (msg.getReason() == (byte) OFPortReason.OFPPR_MODIFY.ordinal()) {
381             updatePhysicalPort(port);
382             //logger.debug("Port " + portNumber + " on " + toString() + " modified");
383         } else if (msg.getReason() == (byte) OFPortReason.OFPPR_ADD.ordinal()) {
384             updatePhysicalPort(port);
385             //logger.debug("Port " + portNumber + " on " + toString() + " added");
386         } else if (msg.getReason() == (byte) OFPortReason.OFPPR_DELETE
387                 .ordinal()) {
388             deletePhysicalPort(port);
389             //logger.debug("Port " + portNumber + " on " + toString() + " deleted");
390         }
391
392     }
393
394     private void startSwitchTimer() {
395         this.periodicTimer = new Timer();
396         this.periodicTimer.scheduleAtFixedRate(new TimerTask() {
397             @Override
398             public void run() {
399                 try {
400                     Long now = System.currentTimeMillis();
401                     if ((now - lastMsgReceivedTimeStamp) > SWITCH_LIVENESS_TIMEOUT) {
402                         if (probeSent) {
403                             // switch failed to respond to our probe, consider it down
404                             logger.warn("{} is idle for too long, disconnect", toString());
405                             reportSwitchStateChange(false);
406                         } else {
407                             // send a probe to see if the switch is still alive
408                             //logger.debug("Send idle probe (Echo Request) to " + switchName());
409                             probeSent = true;
410                             OFMessage echo = factory
411                                     .getMessage(OFType.ECHO_REQUEST);
412                             asyncFastSend(echo);
413                         }
414                     } else {
415                         if (state == SwitchState.WAIT_FEATURES_REPLY) {
416                             // send another features request
417                             OFMessage request = factory
418                                     .getMessage(OFType.FEATURES_REQUEST);
419                             asyncFastSend(request);
420                         } else {
421                             if (state == SwitchState.WAIT_CONFIG_REPLY) {
422                                 //  send another config request
423                                 OFSetConfig config = (OFSetConfig) factory
424                                         .getMessage(OFType.SET_CONFIG);
425                                 config.setMissSendLength((short) 0xffff)
426                                         .setLengthU(OFSetConfig.MINIMUM_LENGTH);
427                                 asyncFastSend(config);
428                                 OFMessage getConfig = factory
429                                         .getMessage(OFType.GET_CONFIG_REQUEST);
430                                 asyncFastSend(getConfig);
431                             }
432                         }
433                     }
434                 } catch (Exception e) {
435                     reportError(e);
436                 }
437             }
438         }, SWITCH_LIVENESS_TIMER, SWITCH_LIVENESS_TIMER);
439     }
440
441     private void cancelSwitchTimer() {
442         if (this.periodicTimer != null) {
443             this.periodicTimer.cancel();
444         }
445     }
446
447     private void reportError(Exception e) {
448         if (e instanceof AsynchronousCloseException) {
449                 logger.debug("Caught exception {}", e.getMessage());
450         } else {
451                 logger.warn("Caught exception {}", e.getMessage());
452         }
453         // notify core of this error event and disconnect the switch
454         ((Controller) core).takeSwitchEventError(this);
455     }
456
457     private void reportSwitchStateChange(boolean added) {
458         if (added) {
459             ((Controller) core).takeSwtichEventAdd(this);
460         } else {
461             ((Controller) core).takeSwitchEventDelete(this);
462         }
463     }
464
465     @Override
466     public Long getId() {
467         return this.sid;
468     }
469
470     private void processFeaturesReply(OFFeaturesReply reply) {
471         if (this.state == SwitchState.WAIT_FEATURES_REPLY) {
472             this.sid = reply.getDatapathId();
473             this.buffers = reply.getBuffers();
474             this.capabilities = reply.getCapabilities();
475             this.tables = reply.getTables();
476             this.actions = reply.getActions();
477             // notify core of this error event
478             for (OFPhysicalPort port : reply.getPorts()) {
479                 updatePhysicalPort(port);
480             }
481             // config the switch to send full data packet
482             OFSetConfig config = (OFSetConfig) factory
483                     .getMessage(OFType.SET_CONFIG);
484             config.setMissSendLength((short) 0xffff).setLengthU(
485                     OFSetConfig.MINIMUM_LENGTH);
486             asyncFastSend(config);
487             // send config request to make sure the switch can handle the set config
488             OFMessage getConfig = factory.getMessage(OFType.GET_CONFIG_REQUEST);
489             asyncFastSend(getConfig);
490             this.state = SwitchState.WAIT_CONFIG_REPLY;
491             // inform core that a new switch is now operational
492             reportSwitchStateChange(true);
493         }
494     }
495
496     private void updatePhysicalPort(OFPhysicalPort port) {
497         Short portNumber = port.getPortNumber();
498         physicalPorts.put(portNumber, port);
499         portBandwidth
500                 .put(
501                         portNumber,
502                         port.getCurrentFeatures()
503                                 & (OFPortFeatures.OFPPF_10MB_FD.getValue()
504                                         | OFPortFeatures.OFPPF_10MB_HD
505                                                 .getValue()
506                                         | OFPortFeatures.OFPPF_100MB_FD
507                                                 .getValue()
508                                         | OFPortFeatures.OFPPF_100MB_HD
509                                                 .getValue()
510                                         | OFPortFeatures.OFPPF_1GB_FD
511                                                 .getValue()
512                                         | OFPortFeatures.OFPPF_1GB_HD
513                                                 .getValue() | OFPortFeatures.OFPPF_10GB_FD
514                                         .getValue()));
515     }
516
517     private void deletePhysicalPort(OFPhysicalPort port) {
518         Short portNumber = port.getPortNumber();
519         physicalPorts.remove(portNumber);
520         portBandwidth.remove(portNumber);
521     }
522
523     @Override
524     public boolean isOperational() {
525         return ((this.state == SwitchState.WAIT_CONFIG_REPLY) || (this.state == SwitchState.OPERATIONAL));
526     }
527
528     @Override
529     public String toString() {
530         return ("["
531                 + this.socket.toString()
532                 + " SWID "
533                 + (isOperational() ? HexString.toHexString(this.sid)
534                         : "unkbown") + "]");
535     }
536
537     @Override
538     public Date getConnectedDate() {
539         return this.connectedDate;
540     }
541
542     public String getInstanceName() {
543         return instanceName;
544     }
545
546     @Override
547     public Object getStatistics(OFStatisticsRequest req) {
548         int xid = getNextXid();
549         StatisticsCollector worker = new StatisticsCollector(this, xid, req);
550         messageWaitingDone.put(xid, worker);
551         Future<Object> submit = executor.submit(worker);
552         Object result = null;
553         try {
554             result = submit
555                     .get(MESSAGE_RESPONSE_TIMER, TimeUnit.MILLISECONDS);
556             return result;
557         } catch (Exception e) {
558             logger.warn("Timeout while waiting for {} replies", req.getType());
559             result = null; // to indicate timeout has occurred
560             return result;
561         }
562     }
563
564     @Override
565     public Object syncSend(OFMessage msg) {
566         Integer xid = getNextXid();
567         SynchronousMessage worker = new SynchronousMessage(this, xid, msg);
568         messageWaitingDone.put(xid, worker);
569         Object result = null;
570         Boolean status = false;
571         Future<Object> submit = executor.submit(worker);
572         try {
573             result = submit
574                     .get(responseTimerValue, TimeUnit.MILLISECONDS);
575             messageWaitingDone.remove(xid);
576             if (result == null) {
577                 // if result  is null, then it means the switch can handle this message successfully
578                 // convert the result into a Boolean with value true
579                 status = true;
580                 //logger.debug("Successfully send " + msg.getType().toString());
581                 result = status;
582             } else {
583                 // if result  is not null, this means the switch can't handle this message
584                 // the result if OFError already
585                 logger.debug("Send {} failed --> {}", 
586                                 msg.getType().toString(), ((OFError) result).toString());
587             }
588             return result;
589         } catch (Exception e) {
590             logger.warn("Timeout while waiting for {} reply", msg.getType().toString());
591             // convert the result into a Boolean with value false
592             status = false;
593             result = status;
594             return result;
595         }
596     }
597
598     /*
599      * Either a BarrierReply or a OFError is received. If this is a reply for an outstanding sync message,
600      * wake up associated task so that it can continue
601      */
602     private void processBarrierReply(OFBarrierReply msg) {
603         Integer xid = msg.getXid();
604         SynchronousMessage worker = (SynchronousMessage) messageWaitingDone
605                 .remove(xid);
606         if (worker == null) {
607             return;
608         }
609         worker.wakeup();
610     }
611
612     private void processErrorReply(OFError errorMsg) {
613         OFMessage offendingMsg = errorMsg.getOffendingMsg();
614         Integer xid;
615         if (offendingMsg != null) {
616             xid = offendingMsg.getXid();
617         } else {
618             xid = errorMsg.getXid();
619         }
620         /*
621          * the error can be a reply to a synchronous message or to a statistic request message
622          */
623         Callable<?> worker = messageWaitingDone.remove(xid);
624         if (worker == null) {
625             return;
626         }
627         if (worker instanceof SynchronousMessage) {
628             ((SynchronousMessage) worker).wakeup(errorMsg);
629         } else {
630             ((StatisticsCollector) worker).wakeup(errorMsg);
631         }
632     }
633
634     private void processStatsReply(OFStatisticsReply reply) {
635         Integer xid = reply.getXid();
636         StatisticsCollector worker = (StatisticsCollector) messageWaitingDone
637                 .get(xid);
638         if (worker == null) {
639             return;
640         }
641         if (worker.collect(reply)) {
642             // if all the stats records are received (collect() returns true)
643             // then we are done.
644             messageWaitingDone.remove(xid);
645             worker.wakeup();
646         }
647     }
648     
649     @Override
650     public Map<Short, OFPhysicalPort> getPhysicalPorts() {
651         return this.physicalPorts;
652     }
653
654     @Override
655     public OFPhysicalPort getPhysicalPort(Short portNumber) {
656         return this.physicalPorts.get(portNumber);
657     }
658
659     @Override
660     public Integer getPortBandwidth(Short portNumber) {
661         return this.portBandwidth.get(portNumber);
662     }
663
664     @Override
665     public Set<Short> getPorts() {
666         return this.physicalPorts.keySet();
667     }
668
669     @Override
670     public Byte getTables() {
671         return this.tables;
672     }
673     
674     @Override
675     public Integer getActions() {
676         return this.actions;
677     }
678     
679     @Override
680     public Integer getCapabilities() {
681         return this.capabilities;
682     }
683
684     @Override
685     public Integer getBuffers() {
686         return this.buffers;
687     }
688
689     @Override
690     public boolean isPortEnabled(short portNumber) {
691         return isPortEnabled(physicalPorts.get(portNumber));
692     }
693
694     @Override
695     public boolean isPortEnabled(OFPhysicalPort port) {
696         if (port == null) {
697             return false;
698         }
699         int portConfig = port.getConfig();
700         int portState = port.getState();
701         if ((portConfig & OFPortConfig.OFPPC_PORT_DOWN.getValue()) > 0) {
702             return false;
703         }
704         if ((portState & OFPortState.OFPPS_LINK_DOWN.getValue()) > 0) {
705             return false;
706         }
707         if ((portState & OFPortState.OFPPS_STP_MASK.getValue()) == OFPortState.OFPPS_STP_BLOCK
708                 .getValue()) {
709             return false;
710         }
711         return true;
712     }
713
714     @Override
715     public List<OFPhysicalPort> getEnabledPorts() {
716         List<OFPhysicalPort> result = new ArrayList<OFPhysicalPort>();
717         synchronized (this.physicalPorts) {
718             for (OFPhysicalPort port : physicalPorts.values()) {
719                 if (isPortEnabled(port)) {
720                     result.add(port);
721                 }
722             }
723         }
724         return result;
725     }
726
727         /*
728          * Transmit thread polls the message out of the priority queue and invokes
729          * messaging service to transmit it over the socket channel
730          */
731     class PriorityMessageTransmit implements Runnable {
732         public void run() {
733             running = true;
734             while (running) {
735                 try {
736                         if (!transmitQ.isEmpty()) {
737                                 PriorityMessage pmsg = transmitQ.poll();
738                                 msgReadWriteService.asyncSend(pmsg.msg);
739                                 logger.trace("Message sent: {}", pmsg.toString());
740                         }
741                         Thread.sleep(10);
742                 } catch (Exception e) {
743                         reportError(e);
744                 }
745             }
746                 transmitQ = null;
747         }
748     }
749
750     /*
751      * Setup and start the transmit thread
752      */
753     private void startTransmitThread() {        
754         this.transmitQ = new PriorityBlockingQueue<PriorityMessage>(11, 
755                                 new Comparator<PriorityMessage>() {
756                                         public int compare(PriorityMessage p1, PriorityMessage p2) {
757                                                 return p2.priority - p1.priority;
758                                         }
759                                 });
760         this.transmitThread = new Thread(new PriorityMessageTransmit());
761         this.transmitThread.start();
762     }
763     
764     /*
765      * Setup communication services
766      */
767     private void setupCommChannel() throws Exception {
768         this.selector = SelectorProvider.provider().openSelector();
769         this.socket.configureBlocking(false);
770         this.socket.socket().setTcpNoDelay(true);        
771         this.msgReadWriteService = getMessageReadWriteService();
772     }
773
774     private void sendFirstHello() {
775         try {
776                 OFMessage msg = factory.getMessage(OFType.HELLO);
777                 asyncFastSend(msg);
778         } catch (Exception e) {
779                 reportError(e);
780         }
781     }
782     
783     private IMessageReadWrite getMessageReadWriteService() throws Exception {
784         String str = System.getProperty("secureChannelEnabled");
785         return ((str != null) && (str.trim().equalsIgnoreCase("true"))) ? 
786                         new SecureMessageReadWriteService(socket, selector) : 
787                         new MessageReadWriteService(socket, selector);
788     }
789 }