package com.juick.xmpp.s2s; import com.juick.xmpp.utils.XmlUtils; import java.io.EOFException; import java.io.IOException; import java.io.InputStreamReader; import java.io.OutputStreamWriter; import java.net.InetSocketAddress; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.Channels; import java.nio.channels.CompletionHandler; import java.util.logging.Level; import org.xmlpull.v1.XmlPullParser; /** * * @author ugnich */ public class ConnectionOut extends Connection { public boolean streamReady = false; public String to; String checkSID = null; String dbKey = null; public ConnectionOut(String hostname) { super(); to = hostname; } public ConnectionOut(String hostname, String checkSID, String dbKey) { super(); to = hostname; this.checkSID = checkSID; this.dbKey = dbKey; } public void parseStream() { LOGGER.info("STREAM TO " + to + " START"); try { HostnamePort addr = DNSQueries.getServerAddress(to); socket = AsynchronousSocketChannel.open(); socket.connect(new InetSocketAddress(addr.hostname, addr.port), socket, new CompletionHandler() { @Override public void completed(Void result, AsynchronousSocketChannel attachment) { try { parser.setInput(new InputStreamReader(Channels.newInputStream(socket))); writer = new OutputStreamWriter(Channels.newOutputStream(socket)); sendStanza(""); parser.next(); // stream:stream streamID = parser.getAttributeValue(null, "id"); if (streamID == null || streamID.isEmpty()) { throw new Exception("STREAM TO " + to + " INVALID FIRST PACKET"); } LOGGER.info("STREAM TO " + to + " " + streamID + " OPEN"); S2SComponent.addConnectionOut(ConnectionOut.this); if (checkSID != null) { sendDialbackVerify(checkSID, dbKey); } sendStanza("" + generateDialbackKey(to, S2SComponent.HOSTNAME, streamID) + ""); while (parser.next() != XmlPullParser.END_DOCUMENT) { if (parser.getEventType() != XmlPullParser.START_TAG) { continue; } logParser(); String tag = parser.getName(); if (tag.equals("db:result")) { String type = parser.getAttributeValue(null, "type"); if (type != null && type.equals("valid")) { streamReady = true; LOGGER.info("STREAM TO " + to + " " + streamID + " READY"); String cache = S2SComponent.getFromCache(to); if (cache != null) { LOGGER.info("STREAM TO " + to + " " + streamID + " SENDING CACHE"); sendStanza(cache); } } else { LOGGER.info("STREAM TO " + to + " " + streamID + " DIALBACK FAIL"); } XmlUtils.skip(parser); } else if (tag.equals("db:verify")) { String from = parser.getAttributeValue(null, "from"); String type = parser.getAttributeValue(null, "type"); String sid = parser.getAttributeValue(null, "id"); if (from != null && from.equals(to) && sid != null && !sid.isEmpty() && type != null) { ConnectionIn c = S2SComponent.getConnectionIn(sid); if (c != null) { c.sendDialbackResult(from, type); } } XmlUtils.skip(parser); } else { LOGGER.info("STREAM TO " + to + " " + streamID + ": " + XmlUtils.parseToString(parser, true)); } } LOGGER.warning("STREAM TO " + to + " " + streamID + " FINISHED"); S2SComponent.removeConnectionOut(ConnectionOut.this); closeConnection(); } catch (EOFException eofex) { LOGGER.info(String.format("STREAM %s %s CLOSED (dirty)", to, streamID)); S2SComponent.removeConnectionOut(ConnectionOut.this); closeConnection(); } catch (Exception e) { LOGGER.log(Level.SEVERE, "s2s out exception", e); S2SComponent.removeConnectionOut(ConnectionOut.this); closeConnection(); } } @Override public void failed(Throwable exc, AsynchronousSocketChannel attachment) { LOGGER.log(Level.WARNING, "s2s out failed", exc); S2SComponent.removeConnectionOut(ConnectionOut.this); closeConnection(); } }); } catch (Exception e) { LOGGER.warning(e.toString()); S2SComponent.removeConnectionOut(this); closeConnection(); } } public void sendDialbackVerify(String sid, String key) { try { sendStanza("" + key + ""); } catch (IOException e) { LOGGER.log(Level.WARNING, "STREAM TO " + to + " " + streamID + " ERROR", e); } } }