PacketReader


1   /**
2    * $RCSfile$
3    * $Revision: 2732 $
4    * $Date: 2005-08-26 23:29:04 -0300 (Fri, 26 Aug 2005) $
5    *
6    * Copyright 2003-2004 Jive Software.
7    *
8    * All rights reserved. Licensed under the Apache License, Version 2.0 (the "License");
9    * you may not use this file except in compliance with the License.
10   * You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  
21  package org.jivesoftware.smack;
22  
23  import org.jivesoftware.smack.filter.PacketFilter;
24  import org.jivesoftware.smack.packet.*;
25  import org.jivesoftware.smack.provider.IQProvider;
26  import org.jivesoftware.smack.provider.ProviderManager;
27  import org.jivesoftware.smack.util.PacketParserUtils;
28  import org.xmlpull.mxp1.MXParser;
29  import org.xmlpull.v1.XmlPullParser;
30  import org.xmlpull.v1.XmlPullParserException;
31  
32  import java.util.*;
33  
34  /**
35   * Listens for XML traffic from the XMPP server and parses it into packet objects.
36   * The packet reader also manages all packet listeners and collectors.<p>
37   *
38   * @see PacketCollector
39   * @see PacketListener
40   * @author Matt Tucker
41   */
42  class PacketReader {
43  
44      private Thread   readerThread;
45      private Thread   listenerThread;
46  
47      private XMPPConnection connection;
48      private XmlPullParser parser;
49      private boolean done = false;
50      protected List collectors = new ArrayList();
51      private List listeners = new ArrayList();
52      protected List connectionListeners = new ArrayList();
53  
54      private String   connectionID = null;
55      private Object   connectionIDLock = new Object  ();
56  
57      protected PacketReader(XMPPConnection connection) {
58          this.connection = connection;
59  
60          readerThread = new Thread  () {
61              public void run() {
62                  parsePackets();
63              }
64          };
65          readerThread.setName("Smack Packet Reader");
66          readerThread.setDaemon(true);
67  
68          listenerThread = new Thread  () {
69              public void run() {
70                  try {
71                      processListeners();
72                  }
73                  catch (Exception   e) {
74                      e.printStackTrace();
75                  }
76              }
77          };
78          listenerThread.setName("Smack Listener Processor");
79          listenerThread.setDaemon(true);
80  
81          try {
82              parser = new MXParser();
83              parser.setFeature(XmlPullParser.FEATURE_PROCESS_NAMESPACES, true);
84              parser.setInput(connection.reader);
85          }
86          catch (XmlPullParserException xppe) {
87              xppe.printStackTrace();
88          }
89      }
90  
91      /**
92       * Creates a new packet collector for this reader. A packet filter determines
93       * which packets will be accumulated by the collector.
94       *
95       * @param packetFilter the packet filter to use.
96       * @return a new packet collector.
97       */
98      public PacketCollector createPacketCollector(PacketFilter packetFilter) {
99          return new PacketCollector(this, packetFilter);
100     }
101 
102     /**
103      * Registers a packet listener with this reader. A packet filter determines
104      * which packets will be delivered to the listener.
105      *
106      * @param packetListener the packet listener to notify of new packets.
107      * @param packetFilter the packet filter to use.
108      */
109     public void addPacketListener(PacketListener packetListener, PacketFilter packetFilter) {
110         ListenerWrapper wrapper = new ListenerWrapper(this, packetListener,
111                 packetFilter);
112         synchronized (listeners) {
113             listeners.add(wrapper);
114         }
115     }
116 
117     /**
118      * Removes a packet listener.
119      *
120      * @param packetListener the packet listener to remove.
121      */
122     public void removePacketListener(PacketListener packetListener) {
123         synchronized (listeners) {
124             for (int i=0; i<listeners.size(); i++) {
125                 ListenerWrapper wrapper = (ListenerWrapper)listeners.get(i);
126                 if (wrapper != null && wrapper.packetListener.equals(packetListener)) {
127                     listeners.set(i, null);
128                 }
129             }
130         }
131     }
132 
133     /**
134      * Starts the packet reader thread and returns once a connection to the server
135      * has been established. A connection will be attempted for a maximum of five
136      * seconds. An XMPPException will be thrown if the connection fails.
137      *
138      * @throws XMPPException if the server fails to send an opening stream back
139      *      for more than five seconds.
140      */
141     public void startup() throws XMPPException {
142         readerThread.start();
143         listenerThread.start();
144         // Wait for stream tag before returing. We'll wait a couple of seconds before
145         // giving up and throwing an error.
146         try {
147             synchronized(connectionIDLock) {
148                 if (connectionID == null) {
149                     // A waiting thread may be woken up before the wait time or a notify
150                     // (although this is a rare thing). Therefore, we continue waiting
151                     // until either a connectionID has been set (and hence a notify was
152                     // made) or the total wait time has elapsed.
153                     long waitTime = SmackConfiguration.getPacketReplyTimeout();
154                     long start = System.currentTimeMillis();
155                     while (connectionID == null && !done) {
156                         if (waitTime <= 0) {
157                             break;
158                         }
159                         // Wait 3 times the standard time since TLS may take a while
160                         connectionIDLock.wait(waitTime * 3);
161                         long now = System.currentTimeMillis();
162                         waitTime -= now - start;
163                         start = now;
164                     }
165                 }
166             }
167         }
168         catch (InterruptedException   ie) { }
169         if (connectionID == null) {
170             throw new XMPPException("Connection failed. No response from server.");
171         }
172         else {
173             connection.connectionID = connectionID;
174         }
175     }
176 
177     /**
178      * Shuts the packet reader down.
179      */
180     public void shutdown() {
181         // Notify connection listeners of the connection closing if done hasn't already been set.
182         if (!done) {
183             ArrayList listenersCopy;
184             synchronized (connectionListeners) {
185                 // Make a copy since it's possible that a listener will be removed from the list
186                 listenersCopy = new ArrayList(connectionListeners);
187                 for (Iterator i=listenersCopy.iterator(); i.hasNext(); ) {
188                     ConnectionListener listener = (ConnectionListener)i.next();
189                     listener.connectionClosed();
190                 }
191             }
192         }
193         done = true;
194     }
195 
196     /**
197      * Sends out a notification that there was an error with the connection
198      * and closes the connection.
199      *
200      * @param e the exception that causes the connection close event.
201      */
202     void notifyConnectionError(Exception   e) {
203         done = true;
204         connection.close();
205         // Print the stack trace to help catch the problem
206         e.printStackTrace();
207         // Notify connection listeners of the error.
208         ArrayList listenersCopy;
209         synchronized (connectionListeners) {
210             // Make a copy since it's possible that a listener will be removed from the list
211             listenersCopy = new ArrayList(connectionListeners);
212             for (Iterator i=listenersCopy.iterator(); i.hasNext(); ) {
213                 ConnectionListener listener = (ConnectionListener)i.next();
214                 listener.connectionClosedOnError(e);
215             }
216         }
217     }
218 
219     /**
220      * Resets the parser using the latest connection's reader. Reseting the parser is necessary
221      * when the plain connection has been secured or when a new opening stream element is going
222      * to be sent by the server.
223      */
224     private void resetParser() throws XmlPullParserException {
225         parser.setInput(connection.reader);
226     }
227 
228     /**
229      * Process listeners.
230      */
231     private void processListeners() {
232         boolean processedPacket = false;
233         while (!done) {
234             synchronized (listeners) {
235                 if (listeners.size() > 0) {
236                     for (int i=listeners.size()-1; i>=0; i--) {
237                         if (listeners.get(i) == null) {
238                             listeners.remove(i);
239                         }
240                     }
241                 }
242             }
243             processedPacket = false;
244             int size = listeners.size();
245             for (int i=0; i<size; i++) {
246                 ListenerWrapper wrapper = (ListenerWrapper)listeners.get(i);
247                 if (wrapper != null) {
248                     processedPacket = processedPacket || wrapper.notifyListener();
249                 }
250             }
251             if (!processedPacket) {
252                 try {
253                     Thread.sleep(100);
254                 }
255                 catch (InterruptedException   ie) { }
256             }
257         }
258     }
259 
260     /**
261      * Parse top-level packets in order to process them further.
262      */
263     private void parsePackets() {
264         try {
265             int eventType = parser.getEventType();
266             do {
267                 if (eventType == XmlPullParser.START_TAG) {
268                     if (parser.getName().equals("message")) {
269                         processPacket(PacketParserUtils.parseMessage(parser));
270                     }
271                     else if (parser.getName().equals("iq")) {
272                         processPacket(parseIQ(parser));
273                     }
274                     else if (parser.getName().equals("presence")) {
275                         processPacket(PacketParserUtils.parsePresence(parser));
276                     }
277                     // We found an opening stream. Record information about it, then notify
278                     // the connectionID lock so that the packet reader startup can finish.
279                     else if (parser.getName().equals("stream")) {
280                         // Ensure the correct jabber:client namespace is being used.
281                         if ("jabber:client".equals(parser.getNamespace(null))) {
282                             // Get the connection id.
283                             for (int i=0; i<parser.getAttributeCount(); i++) {
284                                 if (parser.getAttributeName(i).equals("id")) {
285                                     if (("1.0".equals(parser.getAttributeValue("", "version")) &&
286                                             connection.isUsingTLS()) || (!"1.0".equals(
287                                             parser.getAttributeValue("", "version")))) {
288                                         // Save the connectionID and notify that we've gotten it.
289                                         // Only notify if TLS has been secured or if the server
290                                         // does not support TLS
291                                         synchronized(connectionIDLock) {
292                                             connectionID = parser.getAttributeValue(i);
293                                             connectionIDLock.notifyAll();
294                                         }
295                                     }
296                                 }
297                                 else if (parser.getAttributeName(i).equals("from")) {
298                                     // Use the server name that the server says that it is.
299                                     connection.serviceName = parser.getAttributeValue(i);
300                                 }
301                             }
302                         }
303                     }
304                     else if (parser.getName().equals("features")) {
305                         parseFeatures(parser);
306                     }
307                     else if (parser.getName().equals("proceed")) {
308                         // Secure the connection by negotiating TLS
309                         connection.proceedTLSReceived();
310                         // Reset the state of the parser since a new stream element is going
311                         // to be sent by the server
312                         resetParser();
313                     }
314                     else if (parser.getName().equals("failure")) {
315                         // TLS negotiation has failed so close the connection.
316                         throw new Exception  ("TLS negotiation has failed");
317                     }
318                     else if (parser.getName().equals("challenge")) {
319                         // The server is challenging the SASL authentication made by the client
320                         connection.getSASLAuthentication().challengeReceived(parser.nextText());
321                     }
322                     else if (parser.getName().equals("success")) {
323                         // The SASL authentication with the server was successful. The next step
324                         // will be to bind the resource
325                         connection.getSASLAuthentication().authenticated();
326                         // Reset the state of the parser since a new stream element is going
327                         // to be sent by the server
328                         resetParser();
329                     }
330                 }
331                 else if (eventType == XmlPullParser.END_TAG) {
332                     if (parser.getName().equals("stream")) {
333                         // Close the connection.
334                         connection.close();
335                     }
336                 }
337                 eventType = parser.next();
338             } while (!done && eventType != XmlPullParser.END_DOCUMENT);
339         }
340         catch (Exception   e) {
341             if (!done) {
342                 // Close the connection and notify connection listeners of the
343                 // error.
344                 notifyConnectionError(e);
345             }
346         }
347     }
348 
349     /**
350      * Processes a packet after it's been fully parsed by looping through the installed
351      * packet collectors and listeners and letting them examine the packet to see if
352      * they are a match with the filter.
353      *
354      * @param packet the packet to process.
355      */
356     private void processPacket(Packet packet) {
357         if (packet == null) {
358             return;
359         }
360 
361         // Remove all null values from the collectors list.
362         synchronized (collectors) {
363             for (int i=collectors.size()-1; i>=0; i--) {
364                     if (collectors.get(i) == null) {
365                         collectors.remove(i);
366                     }
367                 }
368         }
369 
370         // Loop through all collectors and notify the appropriate ones.
371         int size = collectors.size();
372         for (int i=0; i<size; i++) {
373             PacketCollector collector = (PacketCollector)collectors.get(i);
374             if (collector != null) {
375                 // Have the collector process the packet to see if it wants to handle it.
376                 collector.processPacket(packet);
377             }
378         }
379     }
380 
381     private void parseFeatures(XmlPullParser parser) throws Exception   {
382         boolean done = false;
383         while (!done) {
384             int eventType = parser.next();
385 
386             if (eventType == XmlPullParser.START_TAG) {
387                 if (parser.getName().equals("starttls")) {
388                     // Confirm the server that we want to use TLS
389                     connection.startTLSReceived();
390                 }
391                 else if (parser.getName().equals("mechanisms") && connection.isUsingTLS()) {
392                     // The server is reporting available SASL mechanism. Store this information
393                     // which will be used later while logging (i.e. authenticating) into
394                     // the server
395                     connection.getSASLAuthentication()
396                             .setAvailableSASLMethods(parseMechanisms(parser));
397                 }
398                 else if (parser.getName().equals("bind")) {
399                     // The server requires the client to bind a resource to the stream
400                     connection.getSASLAuthentication().bindingRequired();
401                 }
402                 else if (parser.getName().equals("session")) {
403                     // The server supports sessions
404                     connection.getSASLAuthentication().sessionsSupported();
405                 }
406             }
407             else if (eventType == XmlPullParser.END_TAG) {
408                 if (parser.getName().equals("features")) {
409                     done = true;
410                 }
411             }
412         }
413     }
414 
415     /**
416      * Returns a collection of Stings with the mechanisms included in the mechanisms stanza.
417      *
418      * @param parser the XML parser, positioned at the start of an IQ packet.
419      * @return a collection of Stings with the mechanisms included in the mechanisms stanza.
420      * @throws Exception if an exception occurs while parsing the stanza.
421      */
422     private Collection parseMechanisms(XmlPullParser parser) throws Exception   {
423         List mechanisms = new ArrayList();
424         boolean done = false;
425         while (!done) {
426             int eventType = parser.next();
427 
428             if (eventType == XmlPullParser.START_TAG) {
429                 String   elementName = parser.getName();
430                 if (elementName.equals("mechanism")) {
431                     mechanisms.add(parser.nextText());
432                 }
433             }
434             else if (eventType == XmlPullParser.END_TAG) {
435                 if (parser.getName().equals("mechanisms")) {
436                     done = true;
437                 }
438             }
439         }
440         return mechanisms;
441     }
442 
443     /**
444      * Parses an IQ packet.
445      *
446      * @param parser the XML parser, positioned at the start of an IQ packet.
447      * @return an IQ object.
448      * @throws Exception if an exception occurs while parsing the packet.
449      */
450     private IQ parseIQ(XmlPullParser parser) throws Exception   {
451         IQ iqPacket = null;
452 
453         String   id = parser.getAttributeValue("", "id");
454         String   to = parser.getAttributeValue("", "to");
455         String   from = parser.getAttributeValue("", "from");
456         IQ.Type type = IQ.Type.fromString(parser.getAttributeValue("", "type"));
457         XMPPError error = null;
458 
459         boolean done = false;
460         while (!done) {
461             int eventType = parser.next();
462 
463             if (eventType == XmlPullParser.START_TAG) {
464                 String   elementName = parser.getName();
465                 String   namespace = parser.getNamespace();
466                 if (elementName.equals("error")) {
467                     error = PacketParserUtils.parseError(parser);
468                 }
469                 else if (elementName.equals("query") && namespace.equals("jabber:iq:auth")) {
470                     iqPacket = parseAuthentication(parser);
471                 }
472                 else if (elementName.equals("query") && namespace.equals("jabber:iq:roster")) {
473                     iqPacket = parseRoster(parser);
474                 }
475                 else if (elementName.equals("query") && namespace.equals("jabber:iq:register")) {
476                     iqPacket = parseRegistration(parser);
477                 }
478                 // Otherwise, see if there is a registered provider for
479                 // this element name and namespace.
480                 else {
481                     Object   provider = ProviderManager.getIQProvider(elementName, namespace);
482                     if (provider != null) {
483                         if (provider instanceof IQProvider) {
484                             iqPacket = ((IQProvider)provider).parseIQ(parser);
485                         }
486                         else if (provider instanceof Class  ) {
487                             iqPacket = (IQ)PacketParserUtils.parseWithIntrospection(elementName,
488                                     (Class  )provider, parser);
489                         }
490                     }
491                 }
492             }
493             else if (eventType == XmlPullParser.END_TAG) {
494                 if (parser.getName().equals("iq")) {
495                     done = true;
496                 }
497             }
498         }
499         // Decide what to do when an IQ packet was not understood
500         if (iqPacket == null) {
501             if (IQ.Type.GET == type || IQ.Type.SET == type ) {
502                 // If the IQ stanza is of type "get" or "set" containing a child element
503                 // qualified by a namespace it does not understand, then answer an IQ of
504                 // type "error" with code 501 ("feature-not-implemented")
505                 iqPacket = new IQ() {
506                     public String   getChildElementXML() {
507                         return null;
508                     }
509                 };
510                 iqPacket.setPacketID(id);
511                 iqPacket.setTo(from);
512                 iqPacket.setFrom(to);
513                 iqPacket.setType(IQ.Type.ERROR);
514                 iqPacket.setError(new XMPPError(501, "feature-not-implemented"));
515                 connection.sendPacket(iqPacket);
516                 return null;
517             }
518             else {
519                 // If an IQ packet wasn't created above, create an empty IQ packet.
520                 iqPacket = new IQ() {
521                     public String   getChildElementXML() {
522                         return null;
523                     }
524                 };
525             }
526         }
527 
528         // Set basic values on the iq packet.
529         iqPacket.setPacketID(id);
530         iqPacket.setTo(to);
531         iqPacket.setFrom(from);
532         iqPacket.setType(type);
533         iqPacket.setError(error);
534 
535         return iqPacket;
536     }
537 
538     private Authentication parseAuthentication(XmlPullParser parser) throws Exception   {
539         Authentication authentication = new Authentication();
540         boolean done = false;
541         while (!done) {
542             int eventType = parser.next();
543             if (eventType == XmlPullParser.START_TAG) {
544                 if (parser.getName().equals("username")) {
545                     authentication.setUsername(parser.nextText());
546                 }
547                 else if (parser.getName().equals("password")) {
548                     authentication.setPassword(parser.nextText());
549                 }
550                 else if (parser.getName().equals("digest")) {
551                     authentication.setDigest(parser.nextText());
552                 }
553                 else if (parser.getName().equals("resource")) {
554                     authentication.setResource(parser.nextText());
555                 }
556             }
557             else if (eventType == XmlPullParser.END_TAG) {
558                 if (parser.getName().equals("query")) {
559                     done = true;
560                 }
561             }
562         }
563         return authentication;
564     }
565 
566     private RosterPacket parseRoster(XmlPullParser parser) throws Exception   {
567         RosterPacket roster = new RosterPacket();
568         boolean done = false;
569         RosterPacket.Item item = null;
570         while (!done) {
571             int eventType = parser.next();
572             if (eventType == XmlPullParser.START_TAG) {
573                 if (parser.getName().equals("item")) {
574                     String   jid = parser.getAttributeValue("", "jid");
575                     String   name = parser.getAttributeValue("", "name");
576                     // Create packet.
577                     item = new RosterPacket.Item(jid, name);
578                     // Set status.
579                     String   ask = parser.getAttributeValue("", "ask");
580                     RosterPacket.ItemStatus status = RosterPacket.ItemStatus.fromString(ask);
581                     item.setItemStatus(status);
582                     // Set type.
583                     String   subscription = parser.getAttributeValue("", "subscription");
584                     RosterPacket.ItemType type = RosterPacket.ItemType.fromString(subscription);
585                     item.setItemType(type);
586                 }
587                 if (parser.getName().equals("group")) {
588                     String   groupName = parser.nextText();
589                     item.addGroupName(groupName);
590                 }
591             }
592             else if (eventType == XmlPullParser.END_TAG) {
593                 if (parser.getName().equals("item")) {
594                     roster.addRosterItem(item);
595                 }
596                 if (parser.getName().equals("query")) {
597                     done = true;
598                 }
599             }
600         }
601         return roster;
602     }
603 
604      private Registration parseRegistration(XmlPullParser parser) throws Exception   {
605         Registration registration = new Registration();
606         Map fields = null;
607         boolean done = false;
608         while (!done) {
609             int eventType = parser.next();
610             if (eventType == XmlPullParser.START_TAG) {
611                 // Any element that's in the jabber:iq:register namespace,
612                 // attempt to parse it if it's in the form <name>value</name>.
613                 if (parser.getNamespace().equals("jabber:iq:register")) {
614                     String   name = parser.getName();
615                     String   value = "";
616                     if (fields == null) {
617                         fields = new HashMap();
618                     }
619 
620                     if (parser.next() == XmlPullParser.TEXT) {
621                         value = parser.getText();
622                     }
623                     // Ignore instructions, but anything else should be added to the map.
624                     if (!name.equals("instructions")) {
625                         fields.put(name, value);
626                     }
627                     else {
628                         registration.setInstructions(value);
629                     }
630 }
631                 // Otherwise, it must be a packet extension.
632                 else {
633                     registration.addExtension(
634                         PacketParserUtils.parsePacketExtension(
635                             parser.getName(),
636                             parser.getNamespace(),
637                             parser));
638                 }
639             }
640             else if (eventType == XmlPullParser.END_TAG) {
641                 if (parser.getName().equals("query")) {
642                     done = true;
643                 }
644             }
645         }
646         registration.setAttributes(fields);
647         return registration;
648     }
649 
650     /**
651      * A wrapper class to associate a packet collector with a listener.
652      */
653     private static class ListenerWrapper {
654 
655         private PacketListener packetListener;
656         private PacketCollector packetCollector;
657 
658         public ListenerWrapper(PacketReader packetReader, PacketListener packetListener,
659                 PacketFilter packetFilter)
660         {
661             this.packetListener = packetListener;
662             this.packetCollector = new PacketCollector(packetReader, packetFilter);
663         }
664 
665         public boolean equals(Object   object) {
666             if (object == null) {
667                 return false;
668             }
669             if (object instanceof ListenerWrapper) {
670                 return ((ListenerWrapper)object).packetListener.equals(this.packetListener);
671             }
672             else if (object instanceof PacketListener) {
673                 return object.equals(this.packetListener);
674             }
675             return false;
676         }
677 
678         public boolean notifyListener() {
679             Packet packet = packetCollector.pollResult();
680             if (packet != null) {
681                 packetListener.processPacket(packet);
682                 return true;
683             }
684             else {
685                 return false;
686             }
687         }
688 
689         public void cancel() {
690             packetCollector.cancel();
691             packetCollector = null;
692             packetListener = null;
693         }
694     }
695 }
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags