xmppclient.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 723 行 · 第 1/2 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * Resin Open Source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.xmpp;

import com.caucho.bam.*;
import com.caucho.server.connection.*;
import com.caucho.server.port.*;
import com.caucho.util.*;
import com.caucho.vfs.*;
import com.caucho.xmpp.im.*;

import java.io.IOException;
import java.io.Serializable;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.logging.*;
import javax.xml.stream.*;

/**
 * XMPP protocol
 */
public class XmppClient {
  private static final L10N L = new L10N(XmppClient.class);
  private static final Logger log
    = Logger.getLogger(XmppClient.class.getName());

  private static final String STREAMS_NS = "http://etherx.jabber.org/streams";

  private static final String AUTH
    = "auth{http://jabber.org/features/iq-auth}";
  private static final String REGISTER
    = "register{http://jabber.org/features/iq-register}";

  private InetAddress _address;
  private int _port;
  
  private String _to;

  private Socket _s;
  private ReadStream _is;
  private WriteStream _os;

  private String _id;
  private String _from;
  
  private XmppStreamReader _in;
  private XmppReader _reader;
  
  private boolean _isFinest;

  private int _mId;

  private HashSet<String> _authMechanisms = new HashSet<String>();
  private HashSet<String> _features = new HashSet<String>();

  private BlockingQueue<Stanza> _stanzaQueue
    = new LinkedBlockingQueue<Stanza>();

  private XmppContext _xmppContext = new XmppContext();
  private XmppClientBrokerStream _toBroker;
  private BindCallback _bindCallback;

  private String _jid;
  
  private BamStream _callback;

  public XmppClient(InetAddress address, int port)
  {
    _address = address;
    _port = port;

    _to = _address.getHostAddress();

    _isFinest = log.isLoggable(Level.FINEST);
  }
					   
  public XmppClient(String address, int port)
  {
    this(getByName(address), port);
    
    _to = address;
  }

  private static InetAddress getByName(String address)
  {
    try {
      return InetAddress.getByName(address);
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  public void connect()
    throws IOException
  {
    try {
      if (_s != null)
	throw new IllegalStateException(L.l("{0} is already connected", this));

      _s = new Socket(_address, _port);

      SocketStream ss = new SocketStream(_s);
    
      _os = new WriteStream(ss);
      _is = new ReadStream(ss);

      _os.print("<?xml version='1.0' encoding='UTF-8' ?>\n");
      _os.setEncoding("utf-8");
      
      startStream();

      _os.flush();
      
      XMLInputFactory factory = XMLInputFactory.newInstance();

      XmppMarshalFactory marshalFactory = new XmppMarshalFactory();
      
      XmppStreamWriterImpl out;
      out = new XmppStreamWriterImpl(_os, marshalFactory);

      XmppWriter writer = new XmppWriter(_xmppContext, out);
      
      _toBroker = new XmppClientBrokerStream(this, writer);
    
      _in = new XmppStreamReaderImpl(_is, marshalFactory);

      _bindCallback = new BindCallback();
      _reader = new XmppReader(_xmppContext, _is, _in, _toBroker,
			       _bindCallback);

      String tag = readStartTag();

      if (! tag.equals("stream")
	  || ! STREAMS_NS.equals(_in.getNamespaceURI())) {
	throw new IOExceptionWrapper(L.l("<{0}> with ns={1} is an unexpected server response",
				  tag, _in.getNamespaceURI()));
      }

      readStreamFeatures();
    } catch (XMLStreamException e) {
      throw new IOExceptionWrapper(e);
    }
  }
  
  public void login(String name, String password)
    throws IOException
  {
    String base64 = Base64.encode("" + (char) 0 + name + (char) 0 + password);

    if (log.isLoggable(Level.FINER))
      log.finer(this + " authenticating " + name);

    _os.print("<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='PLAIN'>");
    _os.print(base64);
    _os.print("</auth>");

    startStream();
    _os.flush();

    try {
      if (! readSuccess())
	throw new RuntimeException("expected success");
      
      if (! readStream())
	throw new RuntimeException("expected stream");

      StringBuilder sb = new StringBuilder();
      Base64.encode(sb, RandomUtil.getRandomLong());
      
      _os.print("<iq type='set' id='" + _mId++ + "'>");
      _os.print("<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'>");
      _os.print("<resource>" + getClass().getSimpleName() + "</resource>");
      _os.print("</bind>");
      _os.print("</iq>");
      _os.flush();

      _reader.readNext();

      if (_jid == null)
	throw new RuntimeException("expected bind");
      
      _os.print("<iq type='set' id='" + _mId++ + "'>");
      _os.print("<session xmlns='urn:ietf:params:xml:ns:xmpp-session'/>");
      _os.print("</iq>");
      _os.flush();

      _reader.readNext();

      /*
      Stanza stanza = _stanzaQueue.poll(2, TimeUnit.SECONDS);

      if (! (stanza instanceof SessionStanza)
	  && ! (stanza instanceof EmptyStanza))
	throw new RuntimeException("expected session");
      */

      if (log.isLoggable(Level.FINER))
	log.finer(this + " authentication successful for " + name);

      _reader.setHandler(_callback);

      ThreadPool.getThreadPool().start(new Listener());
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  public void setCallback(BamStream callback)
  {
    _callback = callback;

    if (_reader != null && _jid != null)
      _reader.setHandler(callback);
  }
  
  public void send(String type, String to, String body)
    throws IOException
  {
    send(type, to, body, null);
  }
  
  public void send(String type, String to, String body, String subject)
    throws IOException
  {
    if (log.isLoggable(Level.FINER))
      log.finer(this + " send to=" + to + " body=" + body);

    try {
      _os.print("<message ");
      _os.print(" type='" + type + "'");
      
      if (to != null)
	_os.print(" to='" + to + "'");
      if (_from != null)
	_os.print(" from='" + _from + "'");
      _os.print(">");
      
      if (subject != null)
	_os.print("<subject>" + subject + "</subject>");
      if (body != null)
	_os.print("<body>" + body + "</body>");
      _os.print("</message>");
      _os.flush();
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  private void startStream()
    throws IOException
  {
    _os.print("<stream:stream");
    _os.print(" to='" + _to + "'");
    _os.print(" xmlns='jabber:client'");
    _os.print(" xmlns:stream='http://etherx.jabber.org/streams'");
    _os.print(" version='1.0'");
    _os.print(">");
    _os.flush();

    if (log.isLoggable(Level.FINER))
      log.finer(this + " <stream:stream xmlns='jabber:client' to='" + _to + "'>");
  }

  private void readStreamFeatures()
    throws IOException, XMLStreamException
  {
    String startTag = readStartTag();

    if (! "features".equals(startTag))
      throw unexpected();

    int tag = 0;
    
    while ((tag = _in.next()) > 0
	   && ! (tag == XMLStreamReader.END_ELEMENT
		 && "features".equals(_in.getLocalName()))) {
      if (_isFinest)
	debug(_in);
      
      if (tag == XMLStreamReader.START_ELEMENT) {
	String localName = _in.getLocalName();

	if ("mechanisms".equals(localName)) {
	}
	else if ("mechanism".equals(localName)) {
	  tag = _in.next();
	  
	  String mechanism = _in.getText();

	  _authMechanisms.add(mechanism);
	}
	else {
	  String feature = localName + "{" + _in.getNamespaceURI() + "}";

	  if (log.isLoggable(Level.FINER))
	    log.finer(this + " feature " + feature);

	  _features.add(feature);
	}
      }
      else if (tag == XMLStreamReader.END_ELEMENT) {
	String localName = _in.getLocalName();
      }
    }
  }

  private String readStartTag()
    throws IOException, XMLStreamException
  {
    int tag = 0;
    
    while ((tag = _in.next()) > 0 && tag != XMLStreamReader.START_ELEMENT) {
      if (_isFinest)
	debug(_in);
    }
    
    if (_isFinest)
      debug(_in);

    return _in.getLocalName();    
  }

  private IOException unexpected()
    throws IOException, XMLStreamException
  {
    if ("error".equals(_in.getLocalName())) {
      int tag;
      
      while ((tag = _in.next()) > 0
	     && ! (tag == XMLStreamReader.END_ELEMENT