- Add stop() method in IMessageReadWrite for proper clean up when connection is closed.
[controller.git] / opendaylight / protocol_plugins / openflow / src / main / java / org / opendaylight / controller / protocol_plugin / openflow / core / internal / Controller.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.SelectionKey;
14 import java.nio.channels.Selector;
15 import java.nio.channels.ServerSocketChannel;
16 import java.nio.channels.SocketChannel;
17 import java.util.Date;
18 import java.util.Iterator;
19 import java.util.Map;
20 import java.util.Map.Entry;
21 import java.util.Set;
22 import java.util.concurrent.BlockingQueue;
23 import java.util.concurrent.ConcurrentHashMap;
24 import java.util.concurrent.ConcurrentMap;
25 import java.util.concurrent.LinkedBlockingQueue;
26 import java.util.concurrent.atomic.AtomicInteger;
27
28 import org.eclipse.osgi.framework.console.CommandInterpreter;
29 import org.eclipse.osgi.framework.console.CommandProvider;
30 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
31 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
32 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
33 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitchStateListener;
34 import org.openflow.protocol.OFMessage;
35 import org.openflow.protocol.OFType;
36 import org.openflow.util.HexString;
37 import org.osgi.framework.BundleContext;
38 import org.osgi.framework.FrameworkUtil;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
41
42 public class Controller implements IController, CommandProvider {
43     private static final Logger logger = LoggerFactory
44             .getLogger(Controller.class);
45     private ControllerIO controllerIO;
46     private Thread switchEventThread;
47     private ConcurrentHashMap<Long, ISwitch> switches;
48     private BlockingQueue<SwitchEvent> switchEvents;
49     // only 1 message listener per OFType
50     private ConcurrentMap<OFType, IMessageListener> messageListeners;
51     // only 1 switch state listener
52     private ISwitchStateListener switchStateListener;
53     private AtomicInteger switchInstanceNumber;
54
55     /*
56      * this thread monitors the switchEvents queue for new incoming events from switch
57      */
58     private class EventHandler implements Runnable {
59         @Override
60         public void run() {
61
62             while (true) {
63                 try {
64                     SwitchEvent ev = switchEvents.take();
65                     SwitchEvent.SwitchEventType eType = ev.getEventType();
66                     ISwitch sw = ev.getSwitch();
67                     if (eType != SwitchEvent.SwitchEventType.SWITCH_MESSAGE) {
68                         //logger.debug("Received " + ev.toString() + " from " + sw.toString());
69                     }
70                     switch (eType) {
71                     case SWITCH_ADD:
72                         Long sid = sw.getId();
73                         ISwitch existingSwitch = switches.get(sid);
74                         if (existingSwitch != null) {
75                             logger.info(" Replacing existing "
76                                     + existingSwitch.toString() + " with New "
77                                     + sw.toString());
78                             disconnectSwitch(existingSwitch);
79                         }
80                         switches.put(sid, sw);
81                         notifySwitchAdded(sw);
82                         break;
83                     case SWITCH_DELETE:
84                         disconnectSwitch(sw);
85                         break;
86                     case SWITCH_ERROR:
87                         disconnectSwitch(sw);
88                         break;
89                     case SWITCH_MESSAGE:
90                         OFMessage msg = ev.getMsg();
91                         if (msg != null) {
92                             IMessageListener listener = messageListeners
93                                     .get(msg.getType());
94                             if (listener != null) {
95                                 listener.receive(sw, msg);
96                             }
97                         }
98                         break;
99                     default:
100                         logger.error("unknow switch event " + eType.ordinal());
101                     }
102                 } catch (InterruptedException e) {
103                     switchEvents.clear();
104                     return;
105                 }
106             }
107         }
108
109     }
110
111     /**
112      * Function called by the dependency manager when all the required
113      * dependencies are satisfied
114      *
115      */
116     public void init() {
117         logger.debug("OpenFlowCore init");
118         this.switches = new ConcurrentHashMap<Long, ISwitch>();
119         this.switchEvents = new LinkedBlockingQueue<SwitchEvent>();
120         this.messageListeners = new ConcurrentHashMap<OFType, IMessageListener>();
121         this.switchStateListener = null;
122         this.switchInstanceNumber = new AtomicInteger(0);
123         registerWithOSGIConsole();
124
125     }
126
127     /**
128      * Function called by dependency manager after "init ()" is called
129      * and after the services provided by the class are registered in
130      * the service registry
131      *
132      */
133     public void start() {
134         logger.debug("OpenFlowCore start() is called");
135         /*
136          * start a thread to handle event coming from the switch
137          */
138         switchEventThread = new Thread(new EventHandler(), "SwitchEvent Thread");
139         switchEventThread.start();
140
141         // spawn a thread to start to listen on the open flow port
142         controllerIO = new ControllerIO(this);
143         try {
144             controllerIO.start();
145         } catch (IOException ex) {
146             logger.error("Caught exception: " + ex + " during start");
147         }
148     }
149
150     /**
151      * Function called by the dependency manager before the services
152      * exported by the component are unregistered, this will be
153      * followed by a "destroy ()" calls
154      *
155      */
156     public void stop() {
157         for (Iterator<Entry<Long, ISwitch>> it = switches.entrySet().iterator(); it
158                 .hasNext();) {
159             Entry<Long, ISwitch> entry = it.next();
160             ((SwitchHandler) entry.getValue()).stop();
161             it.remove();
162         }
163         switchEventThread.interrupt();
164         try {
165                 controllerIO.shutDown();
166         } catch (IOException ex) {
167                 logger.error("Caught exception: " + ex + " during stop");
168         }
169     }
170
171     /**
172      * Function called by the dependency manager when at least one
173      * dependency become unsatisfied or when the component is shutting
174      * down because for example bundle is being stopped.
175      *
176      */
177     public void destroy() {
178     }
179
180     @Override
181     public void addMessageListener(OFType type, IMessageListener listener) {
182         IMessageListener currentListener = this.messageListeners.get(type);
183         if (currentListener != null) {
184             logger.warn(type.toString() + " already listened by "
185                     + currentListener.toString());
186         }
187         this.messageListeners.put(type, listener);
188         logger.debug(type.toString() + " is now listened by "
189                 + listener.toString());
190     }
191
192     @Override
193     public void removeMessageListener(OFType type, IMessageListener listener) {
194         IMessageListener currentListener = this.messageListeners.get(type);
195         if ((currentListener != null) && (currentListener == listener)) {
196             this.messageListeners.remove(type);
197         }
198     }
199
200     @Override
201     public void addSwitchStateListener(ISwitchStateListener listener) {
202         if (this.switchStateListener != null) {
203             logger.warn(this.switchStateListener.toString()
204                     + "already listened to switch events");
205         }
206         this.switchStateListener = listener;
207         logger.debug(listener.toString() + " now listens to switch events");
208     }
209
210     @Override
211     public void removeSwitchStateListener(ISwitchStateListener listener) {
212         if ((this.switchStateListener != null)
213                 && (this.switchStateListener == listener)) {
214             this.switchStateListener = null;
215         }
216     }
217
218     public void handleNewConnection(Selector selector,
219             SelectionKey serverSelectionKey) {
220         ServerSocketChannel ssc = (ServerSocketChannel) serverSelectionKey
221                 .channel();
222         SocketChannel sc = null;
223         try {
224             sc = ssc.accept();
225             // create new switch
226             int i = this.switchInstanceNumber.addAndGet(1);
227             String instanceName = "SwitchHandler-" + i;
228             SwitchHandler switchHandler = new SwitchHandler(this, sc,
229                     instanceName);
230             switchHandler.start();
231             logger.info(instanceName + " connected: " + sc.toString());
232         } catch (IOException e) {
233             return;
234         }
235     }
236
237     private void disconnectSwitch(ISwitch sw) {
238         if (((SwitchHandler) sw).isOperational()) {
239             Long sid = sw.getId();
240             if (this.switches.remove(sid, sw)) {
241                 logger.warn(sw.toString() + " is disconnected");
242                 notifySwitchDeleted(sw);
243             } else {
244                 //logger.warn(sw.toString() + " has been replaced by " +
245                 //      this.switches.get(sid));
246             }
247         }
248         ((SwitchHandler) sw).stop();
249         sw = null;
250     }
251
252     private void notifySwitchAdded(ISwitch sw) {
253         if (switchStateListener != null) {
254             switchStateListener.switchAdded(sw);
255         }
256     }
257
258     private void notifySwitchDeleted(ISwitch sw) {
259         if (switchStateListener != null) {
260             switchStateListener.switchDeleted(sw);
261         }
262     }
263
264     private synchronized void addSwitchEvent(SwitchEvent event) {
265         try {
266             this.switchEvents.put(event);
267         } catch (InterruptedException e) {
268             logger.debug("SwitchEvent caught Interrupt Exception");
269         }
270     }
271
272     public void takeSwtichEventAdd(ISwitch sw) {
273         SwitchEvent ev = new SwitchEvent(
274                 SwitchEvent.SwitchEventType.SWITCH_ADD, sw, null);
275         addSwitchEvent(ev);
276     }
277
278     public void takeSwitchEventDelete(ISwitch sw) {
279         SwitchEvent ev = new SwitchEvent(
280                 SwitchEvent.SwitchEventType.SWITCH_DELETE, sw, null);
281         addSwitchEvent(ev);
282     }
283
284     public void takeSwitchEventError(ISwitch sw) {
285         SwitchEvent ev = new SwitchEvent(
286                 SwitchEvent.SwitchEventType.SWITCH_ERROR, sw, null);
287         addSwitchEvent(ev);
288     }
289
290     public void takeSwitchEventMsg(ISwitch sw, OFMessage msg) {
291         if (messageListeners.get(msg.getType()) != null) {
292             SwitchEvent ev = new SwitchEvent(
293                     SwitchEvent.SwitchEventType.SWITCH_MESSAGE, sw, msg);
294             addSwitchEvent(ev);
295         }
296     }
297
298     @Override
299     public Map<Long, ISwitch> getSwitches() {
300         return this.switches;
301     }
302
303     @Override
304     public ISwitch getSwitch(Long switchId) {
305         return this.switches.get(switchId);
306     }
307
308     public void _controllerShowSwitches(CommandInterpreter ci) {
309         Set<Long> sids = switches.keySet();
310         StringBuffer s = new StringBuffer();
311         int size = sids.size();
312         if (size == 0) {
313             ci.print("switches: empty");
314             return;
315         }
316         Iterator<Long> iter = sids.iterator();
317         s.append("Total: " + size + " switches\n");
318         while (iter.hasNext()) {
319             Long sid = iter.next();
320             Date date = switches.get(sid).getConnectedDate();
321             String switchInstanceName = ((SwitchHandler) switches.get(sid)).getInstanceName();
322             s.append(switchInstanceName + "/" + HexString.toHexString(sid)
323                     + " connected since " + date.toString() + "\n");
324         }
325         ci.print(s.toString());
326         return;
327     }
328
329     public void _controllerReset(CommandInterpreter ci) {
330         ci.print("...Disconnecting the communication to all switches...\n");
331         stop();
332         try {
333             Thread.sleep(1000);
334         } catch (InterruptedException ie) {
335         } finally {
336             ci.print("...start to accept connections from switches...\n");
337             start();
338         }
339     }
340
341     private void registerWithOSGIConsole() {
342         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
343                 .getBundleContext();
344         bundleContext.registerService(CommandProvider.class.getName(), this,
345                 null);
346     }
347
348     @Override
349     public String getHelp() {
350         StringBuffer help = new StringBuffer();
351         help.append("--Open Flow Controller --\n");
352         help.append("\tcontrollerShowSwitches\n");
353         help.append("\tcontrollerReset\n");
354         return help.toString();
355     }
356 }