o More common XML parsing and HTTP client code. o Reimplemented the Nexus code to consume the timeline instead of calculating its own diff.

5 files changed, 396 insertions, 0 deletions

diff --git a/src/main/java/io/trygvis/esper/testing/util/EasySSLSocketFactory.java b/src/main/java/io/trygvis/esper/testing/util/EasySSLSocketFactory.java
new file mode 100755
index 0000000..c1f1a7e
--- /dev/null
+++ b/src/main/java/io/trygvis/esper/testing/util/EasySSLSocketFactory.java
@@ -0,0 +1,104 @@
+package io.trygvis.esper.testing.util;
+
+/*
+ * ====================================================================
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+import java.io.*;
+import java.net.*;
+import javax.net.ssl.*;
+import org.apache.http.conn.*;
+import org.apache.http.conn.scheme.*;
+import org.apache.http.params.*;
+
+public class EasySSLSocketFactory implements SocketFactory, LayeredSocketFactory {
+
+    private SSLContext sslcontext = null;
+
+    private static SSLContext createEasySSLContext() throws IOException {
+        try {
+            SSLContext context = SSLContext.getInstance("TLS");
+            context.init(null, new TrustManager[]{new EasyX509TrustManager(null)}, null);
+            return context;
+        } catch (Exception e) {
+            throw new IOException(e.getMessage());
+        }
+    }
+
+    private SSLContext getSSLContext() throws IOException {
+        if (this.sslcontext == null) {
+            this.sslcontext = createEasySSLContext();
+        }
+        return this.sslcontext;
+    }
+
+    public Socket connectSocket(Socket sock, String host, int port, InetAddress localAddress, int localPort,
+                                HttpParams params) throws IOException, ConnectTimeoutException {
+        int connTimeout = HttpConnectionParams.getConnectionTimeout(params);
+        int soTimeout = HttpConnectionParams.getSoTimeout(params);
+        InetSocketAddress remoteAddress = new InetSocketAddress(host, port);
+        SSLSocket sslSocket = (SSLSocket) ((sock != null) ? sock : createSocket());
+
+        if ((localAddress != null) || (localPort > 0)) {
+            // we need to bind explicitly
+            if (localPort < 0) {
+                localPort = 0; // indicates "any"
+            }
+            InetSocketAddress isa = new InetSocketAddress(localAddress, localPort);
+            sslSocket.bind(isa);
+        }
+
+        sslSocket.connect(remoteAddress, connTimeout);
+        sslSocket.setSoTimeout(soTimeout);
+
+        return sslSocket;
+    }
+
+    public Socket createSocket() throws IOException {
+        return getSSLContext().getSocketFactory().createSocket();
+    }
+
+    public boolean isSecure(Socket socket) throws IllegalArgumentException {
+        return true;
+    }
+
+    public Socket createSocket(Socket socket, String host, int port, boolean autoClose) throws IOException {
+        return getSSLContext().getSocketFactory().createSocket(socket, host, port, autoClose);
+    }
+
+    // -------------------------------------------------------------------
+    // javadoc in org.apache.http.conn.scheme.SocketFactory says :
+    // Both Object.equals() and Object.hashCode() must be overridden
+    // for the correct operation of some connection managers
+    // -------------------------------------------------------------------
+
+    public boolean equals(Object obj) {
+        return ((obj != null) && obj.getClass().equals(EasySSLSocketFactory.class));
+    }
+
+    public int hashCode() {
+        return EasySSLSocketFactory.class.hashCode();
+    }
+}
diff --git a/src/main/java/io/trygvis/esper/testing/util/EasyX509TrustManager.java b/src/main/java/io/trygvis/esper/testing/util/EasyX509TrustManager.java
new file mode 100755
index 0000000..5a4c4d3
--- /dev/null
+++ b/src/main/java/io/trygvis/esper/testing/util/EasyX509TrustManager.java
@@ -0,0 +1,76 @@
+package io.trygvis.esper.testing.util;
+
+/*
+ * ====================================================================
+ *
+ *  Licensed to the Apache Software Foundation (ASF) under one or more
+ *  contributor license agreements.  See the NOTICE file distributed with
+ *  this work for additional information regarding copyright ownership.
+ *  The ASF licenses this file to You under the Apache License, Version 2.0
+ *  (the "License"); you may not use this file except in compliance with
+ *  the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.cert.CertificateException;
+import java.security.cert.X509Certificate;
+
+
+public class EasyX509TrustManager implements X509TrustManager
+{
+    private X509TrustManager standardTrustManager = null;
+
+    public EasyX509TrustManager(KeyStore keystore) throws NoSuchAlgorithmException, KeyStoreException {
+        super();
+        TrustManagerFactory factory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+        factory.init(keystore);
+        TrustManager[] trustmanagers = factory.getTrustManagers();
+        if (trustmanagers.length == 0) {
+            throw new NoSuchAlgorithmException("no trust manager found");
+        }
+        this.standardTrustManager = (X509TrustManager)trustmanagers[0];
+    }
+
+    public void checkClientTrusted(X509Certificate[] certificates,String authType) throws CertificateException {
+        standardTrustManager.checkClientTrusted(certificates,authType);
+    }
+
+    public void checkServerTrusted(X509Certificate[] certificates,String authType) throws CertificateException {
+//        if (certificates != null) {
+//            System.out.println("Server certificate chain:");
+//            for (int i = 0; i < certificates.length; i++) {
+//                System.out.println("X509Certificate[" + i + "]=" + certificates[i]);
+//            }
+//        }
+
+        if ((certificates != null) && (certificates.length == 1)) {
+            certificates[0].checkValidity();
+        } else {
+            standardTrustManager.checkServerTrusted(certificates,authType);
+        }
+    }
+
+    public X509Certificate[] getAcceptedIssuers() {
+        return this.standardTrustManager.getAcceptedIssuers();
+    }
+}
diff --git a/src/main/java/io/trygvis/esper/testing/util/HttpClient.java b/src/main/java/io/trygvis/esper/testing/util/HttpClient.java
new file mode 100755
index 0000000..ef6f2a9
--- /dev/null
+++ b/src/main/java/io/trygvis/esper/testing/util/HttpClient.java
@@ -0,0 +1,123 @@
+package io.trygvis.esper.testing.util;
+
+import fj.*;
+import fj.data.*;
+import io.trygvis.esper.testing.*;
+import static java.lang.System.*;
+
+import org.apache.http.conn.scheme.*;
+import org.apache.http.impl.client.*;
+import org.apache.http.impl.conn.tsccm.*;
+import org.apache.http.params.*;
+import org.codehaus.httpcache4j.*;
+import org.codehaus.httpcache4j.cache.*;
+import org.codehaus.httpcache4j.resolver.*;
+import org.jdom2.*;
+
+import java.io.*;
+import java.net.*;
+import javax.xml.stream.*;
+
+public class HttpClient<A> {
+
+    private final HTTPCache http;
+    private final F<InputStream, Option<A>> f;
+
+    public HttpClient(HTTPCache http, F<InputStream, Option<A>> f) {
+        this.http = http;
+        this.f = f;
+    }
+
+    public Option<A> fetch(URI uri) throws IOException {
+        HTTPResponse response = null;
+
+        try {
+            response = http.execute(new HTTPRequest(uri));
+
+            if (response.getStatus().getCode() != 200) {
+                throw new IOException("Did not get 200 back, got " + response.getStatus().getCode());
+            }
+
+//            return getDocument(response.getPayload().getInputStream());
+            return f.f(response.getPayload().getInputStream());
+        } catch (HTTPException e) {
+            throw new IOException(e);
+        } finally {
+            if (response != null) {
+                response.consume();
+            }
+        }
+    }
+
+
+    public static HTTPCache createHttpCache(Config config) {
+        SchemeRegistry schemeRegistry = new SchemeRegistry();
+        schemeRegistry.register(new Scheme("http", PlainSocketFactory.getSocketFactory(), 80));
+        schemeRegistry.register(new Scheme("https", new EasySSLSocketFactory(), 443));
+        BasicHttpParams params = new BasicHttpParams();
+        ThreadSafeClientConnManager cm = new ThreadSafeClientConnManager(params, schemeRegistry);
+        ResponseResolver responseResolver = new HTTPClientResponseResolver(new DefaultHttpClient(cm, new BasicHttpParams()));
+
+        if (config.gitorious.sessionValue.isSome()) {
+            responseResolver = new GitoriousResponseResolver(config.gitorious.sessionValue.some(), responseResolver);
+        }
+
+        responseResolver = new TimingResponseResolver(responseResolver);
+
+        return new HTTPCache(new MemoryCacheStorage(), responseResolver);
+    }
+
+    private static class TimingResponseResolver implements ResponseResolver {
+        private final ResponseResolver r;
+
+        private TimingResponseResolver(ResponseResolver r) {
+            this.r = r;
+        }
+
+        public HTTPResponse resolve(HTTPRequest request) throws IOException {
+            System.out.println(request.getRequestURI() + ": Executing");
+            long start = currentTimeMillis();
+            Status status = null;
+            try {
+                HTTPResponse response = r.resolve(request);
+                status = response.getStatus();
+                return response;
+            } finally {
+                long end = currentTimeMillis();
+
+                String s = request.getRequestURI() + ": Executed in " + (end - start) + "ms, ";
+
+                if (status != null) {
+                    s += "response: " + status.getCode() + " " + status.getName();
+                } else {
+                    s += "with exception";
+                }
+
+                System.out.println(s);
+            }
+        }
+
+        public void shutdown() {
+            r.shutdown();
+        }
+    }
+
+    private static class GitoriousResponseResolver implements ResponseResolver {
+        private final String gitoriousSessionValue;
+        private final ResponseResolver responseResolver;
+
+        public GitoriousResponseResolver(String gitoriousSessionValue, ResponseResolver responseResolver) {
+            this.gitoriousSessionValue = gitoriousSessionValue;
+            this.responseResolver = responseResolver;
+        }
+
+        public HTTPResponse resolve(HTTPRequest request) throws IOException {
+            request = request.addHeader("Cookie", "_gitorious_sess=" + gitoriousSessionValue);
+            return responseResolver.resolve(request);
+        }
+
+        public void shutdown() {
+            responseResolver.shutdown();
+        }
+    }
+}
diff --git a/src/main/java/io/trygvis/esper/testing/util/XmlHttpClient.java b/src/main/java/io/trygvis/esper/testing/util/XmlHttpClient.java
new file mode 100755
index 0000000..2c2f922
--- /dev/null
+++ b/src/main/java/io/trygvis/esper/testing/util/XmlHttpClient.java
@@ -0,0 +1,27 @@
+package io.trygvis.esper.testing.util;
+
+import fj.*;
+import fj.data.*;
+import org.codehaus.httpcache4j.cache.*;
+import org.jdom2.*;
+
+import java.io.*;
+import java.net.*;
+
+public class XmlHttpClient {
+
+    private final HttpClient<Document> httpClient;
+
+    public XmlHttpClient(HTTPCache http) {
+        final XmlParser parser = new XmlParser();
+        httpClient = new HttpClient<>(http, new F<InputStream, Option<Document>>() {
+            public Option<Document> f(InputStream inputStream) {
+                return parser.parseDocument(inputStream);
+            }
+        });
+    }
+
+    public Option<Document> fetch(URI uri) throws IOException {
+        return httpClient.fetch(uri);
+    }
+}
diff --git a/src/main/java/io/trygvis/esper/testing/util/XmlParser.java b/src/main/java/io/trygvis/esper/testing/util/XmlParser.java
new file mode 100755
index 0000000..e1b463f
--- /dev/null
+++ b/src/main/java/io/trygvis/esper/testing/util/XmlParser.java
@@ -0,0 +1,66 @@
+package io.trygvis.esper.testing.util;
+
+import fj.data.*;
+import static fj.data.Option.*;
+import static javax.xml.stream.XMLStreamConstants.*;
+import org.h2.util.*;
+import org.jdom2.*;
+import org.jdom2.input.*;
+
+import java.io.*;
+import javax.xml.stream.*;
+
+public class XmlParser {
+    public static boolean debugXml;
+
+    private final XMLInputFactory xmlInputFactory;
+
+    private final StAXStreamBuilder streamBuilder = new StAXStreamBuilder();
+
+    public XmlParser() {
+        xmlInputFactory = XMLInputFactory.newFactory();
+    }
+
+    public Option<Document> parseDocument(InputStream stream) {
+
+        try {
+            if (debugXml) {
+                // TODO: Pretty print
+                ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+                IOUtils.copy(stream, buffer);
+                byte[] bytes = buffer.toByteArray();
+                System.out.println("------------------------------------------------");
+                System.out.write(bytes);
+                System.out.println();
+                System.out.println("------------------------------------------------");
+                stream = new ByteArrayInputStream(bytes);
+            }
+
+            // https://github.com/hunterhacker/jdom/issues/101
+            XMLStreamReader readerX = xmlInputFactory.createXMLStreamReader(stream);
+
+            XMLStreamReader reader = xmlInputFactory.createFilteredReader(readerX, new StreamFilter() {
+
+                boolean seenStartDocument;
+
+                @Override
+                public boolean accept(XMLStreamReader reader) {
+                    if(reader.getEventType() == SPACE && !seenStartDocument) {
+                        return false;
+                    }
+
+                    if(reader.getEventType() == START_DOCUMENT) {
+                        seenStartDocument = false;
+                    }
+
+                    return true;
+                }
+            });
+
+            return some(streamBuilder.build(reader));
+        } catch (Exception e) {
+            e.printStackTrace();
+            return none();
+        }
+    }
+}