Per-client CXF bus & shut down on disconnect

client/src/main/java/io/cloudsoft/winrm4j/client/CliClient.java

@@ -1,7 +1,5 @@
 package io.cloudsoft.winrm4j.client;
 
-import org.apache.cxf.bus.CXFBusFactory;
-
 import java.io.OutputStreamWriter;
 import java.io.PrintWriter;
 
@@ -15,7 +13,7 @@ public static void main(String[] args) throws Exception{
 //        System.setProperty("com.sun.xml.internal.ws.transport.http.HttpAdapter.dumpTreshold", Integer.toString(Integer.MAX_VALUE));
 
         if (args.length != 4) {
-            System.out.println("Usage: CliClient <endpoing> <username> <password> <command>");
+            System.out.println("Usage: CliClient <endpoint> <username> <password> <command>");
         }
 
         String endpoint = args[0];
@@ -35,8 +33,6 @@ public static void main(String[] args) throws Exception{
             exitCode = client.command(cmd, new PrintWriter(new OutputStreamWriter(System.out)), new PrintWriter(new OutputStreamWriter(System.err)));
         } finally {
             client.disconnect();
-
-            CXFBusFactory.clearDefaultBusForAnyThread(CXFBusFactory.getDefaultBus());
         }
         System.exit(exitCode);
     }

client/src/main/java/io/cloudsoft/winrm4j/client/WinRmClient.java

@@ -27,6 +27,7 @@
 import javax.xml.ws.spi.ServiceDelegate;
 
 import org.apache.cxf.Bus;
+import org.apache.cxf.Bus.BusState;
 import org.apache.cxf.BusFactory;
 import org.apache.cxf.configuration.jsse.TLSClientParameters;
 import org.apache.cxf.endpoint.Client;
@@ -36,7 +37,6 @@
 import org.apache.cxf.jaxws.spi.ProviderImpl;
 import org.apache.cxf.service.model.ServiceInfo;
 import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduit;
-import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory;
 import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
 import org.apache.cxf.ws.addressing.AddressingProperties;
 import org.apache.cxf.ws.addressing.AttributedURIType;
@@ -112,6 +112,9 @@ public class WinRmClient {
     private Integer retriesForConnectionFailures;
     private Map<String, String> environment;
 
+    private WinRmClientContext context;
+    private boolean cleanupContext;
+
     private WinRm winrm;
     private String shellId;
     private SelectorSetType shellSelector;
@@ -121,6 +124,7 @@ public class WinRmClient {
     private boolean disableCertificateChecks;
     private HostnameVerifier hostnameVerifier;
 
+
     public static Builder builder(URL endpoint) {
         return new Builder(endpoint, AuthSchemes.NTLM);
     }
@@ -141,6 +145,7 @@ public static class Builder {
         private static final java.util.Locale DEFAULT_LOCALE = java.util.Locale.US;
         public static final Long DEFAULT_OPERATION_TIMEOUT = 60l * 1000l;
         private WinRmClient client;
+
         public Builder(URL endpoint, String authenticationScheme) {
             client = new WinRmClient(endpoint, authenticationScheme);
         }
@@ -203,6 +208,11 @@ public Builder hostnameVerifier(HostnameVerifier hostnameVerifier) {
         	return this;
         }
 
+        public Builder context(WinRmClientContext context) {
+            client.context = context;
+            return this;
+        }
+
         public WinRmClient build() {
             if (client.locale == null) {
                 locale(DEFAULT_LOCALE);
@@ -233,14 +243,6 @@ private static String toDuration(long operationTimeout) {
     private WinRmClient(URL endpoint, String authenticationScheme) {
         this.authenticationScheme = authenticationScheme != null ? authenticationScheme : AuthSchemes.NTLM;
         this.endpoint = endpoint;
-
-        if (!this.authenticationScheme.equals(AuthSchemes.BASIC)) {
-            // TODO consider using async client for Basic authentication
-            // Needed to be async according to http://cxf.apache.org/docs/asynchronous-client-http-transport.html
-            Bus bus = BusFactory.getDefaultBus();
-            bus.getProperties().put(AsyncHTTPConduit.USE_ASYNC, Boolean.TRUE);
-            bus.getProperties().put(AsyncHTTPConduitFactory.USE_POLICY, "ALWAYS");
-        }
     }
 
     public int command(String cmd, Writer out, Writer err) {
@@ -416,11 +418,27 @@ private void getStreams(ReceiveResponse receiveResponse, Writer out, Writer err)
         }
     }
 
-    private WinRm getService() {
+    private synchronized WinRm getService() {
         if (winrm != null) {
             return winrm;
         } else {
-            return createService();
+
+            if (context != null) {
+                cleanupContext = false;
+            } else {
+                context = WinRmClientContext.newInstance();
+                cleanupContext = true;
+            }
+
+            Bus prevBus = BusFactory.getAndSetThreadDefaultBus(context.getBus());
+            try {
+                // The default thread bus is set on the ClientImpl and used for further requests
+                return createService();
+            } finally {
+                if (BusFactory.getThreadDefaultBus(false) != prevBus) {
+                    BusFactory.setThreadDefaultBus(prevBus);
+                }
+            }
         }
     }
 
@@ -494,6 +512,7 @@ private synchronized void doCreateServiceNormal() {
     }
 
     //  sys prop approach
+    @SuppressWarnings("unused")
     private void doCreateServiceWithSystemPropertySet() {
         System.setProperty("javax.xml.ws.spi.Provider", ProviderImpl.class.getName());
         doCreateServiceNormal();
@@ -532,7 +551,8 @@ private synchronized Client doCreateServiceWithBean_Part1() {
         factory.getClientFactoryBean().getServiceFactory().setWsdlURL(WinRmService.WSDL_LOCATION);
         factory.setServiceName(WinRmService.SERVICE);
         factory.setEndpointName(WinRmService.WinRmPort);
-        factory.setFeatures(Arrays.asList((Feature)newMemberSubmissionAddressingFeature()));        
+        factory.setFeatures(Arrays.asList((Feature)newMemberSubmissionAddressingFeature()));
+        factory.setBus(context.getBus());
         winrm = factory.create(WinRm.class);
 
         return ClientProxy.getClient(winrm);
@@ -541,6 +561,7 @@ private synchronized Client doCreateServiceWithBean_Part1() {
 
     // approach using CCL
 
+    @SuppressWarnings("unused")
     private synchronized void doCreateServiceInSpecialClassLoader(ClassLoader cl) {
 
         ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
@@ -681,7 +702,7 @@ public X509Certificate[] getAcceptedIssuers() {
         optSetCreate.getOption().add(optCodepage);
 
         final Holder<Shell> holder = new Holder<>(shell);
-        ResourceCreated resourceCreated = winrmCallRetryConnFailure(new CallableFunction<ResourceCreated>() {
+        winrmCallRetryConnFailure(new CallableFunction<ResourceCreated>() {
             @Override
             public ResourceCreated call() {
                 //TODO use different instances of service http://cxf.apache.org/docs/developing-a-consumer.html#DevelopingaConsumer-SettingConnectionPropertieswithContexts
@@ -723,18 +744,28 @@ private static WebServiceFeature newMemberSubmissionAddressingFeature() {
     }
 
     public void disconnect() {
-        if (winrm != null && shellSelector != null) {
-            try {
-                winrmCallRetryConnFailure(new CallableFunction<DeleteResponse>() {
-                    @Override
-                    public DeleteResponse call() {
-                        //TODO use different instances of service http://cxf.apache.org/docs/developing-a-consumer.html#DevelopingaConsumer-SettingConnectionPropertieswithContexts
-                        setActionToContext((BindingProvider) winrm, "http://schemas.xmlsoap.org/ws/2004/09/transfer/Delete");
-                        return winrm.delete(null, RESOURCE_URI, MAX_ENVELOPER_SIZE, operationTimeout, locale, shellSelector);
-                    }
-                });
-            } catch (SOAPFaultException soapFault) {
-                assertFaultCode(soapFault, WSMAN_FAULT_CODE_SHELL_WAS_NOT_FOUND);
+        if (context == null) return;
+        boolean isBusRunning = context.getBus().getState() != BusState.SHUTDOWN;
+        if (!isBusRunning) return;
+        try {
+            if (winrm != null && shellSelector != null) {
+                try {
+                    winrmCallRetryConnFailure(new CallableFunction<DeleteResponse>() {
+                        @Override
+                        public DeleteResponse call() {
+                            //TODO use different instances of service http://cxf.apache.org/docs/developing-a-consumer.html#DevelopingaConsumer-SettingConnectionPropertieswithContexts
+                            setActionToContext((BindingProvider) winrm, "http://schemas.xmlsoap.org/ws/2004/09/transfer/Delete");
+                            return winrm.delete(null, RESOURCE_URI, MAX_ENVELOPER_SIZE, operationTimeout, locale, shellSelector);
+                        }
+                    });
+                } catch (SOAPFaultException soapFault) {
+                    assertFaultCode(soapFault, WSMAN_FAULT_CODE_SHELL_WAS_NOT_FOUND);
+                }
+            }
+        } finally {
+            if (cleanupContext) {
+                context.getBus().shutdown(true);
+                context = null;
             }
         }
     }

client/src/main/java/io/cloudsoft/winrm4j/client/WinRmClientContext.java

@@ -0,0 +1,56 @@
+/*
+ * 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.
+ */
+package io.cloudsoft.winrm4j.client;
+
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduit;
+import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory;
+import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduitFactory.UseAsyncPolicy;
+
+public class WinRmClientContext {
+    public static WinRmClientContext newInstance() {
+        Bus bus = configureBus(BusFactory.newInstance().createBus());
+        return new WinRmClientContext(bus);
+    }
+
+    static Bus configureBus(Bus bus) {
+        // Needed to be async to force the use of Apache HTTP Components client.
+        // Details at http://cxf.apache.org/docs/asynchronous-client-http-transport.html.
+        // Apache HTTP Components needed to support NTLM authentication.
+        bus.getProperties().put(AsyncHTTPConduit.USE_ASYNC, Boolean.TRUE);
+        bus.getProperties().put(AsyncHTTPConduitFactory.USE_POLICY, UseAsyncPolicy.ALWAYS);
+        return bus;
+    }
+
+    private final Bus bus;
+
+    private WinRmClientContext(Bus bus) {
+        this.bus = bus;
+    }
+
+    public void shutdown() {
+        bus.shutdown(true);
+    }
+
+    Bus getBus() {
+        return bus;
+    }
+
+}

winrm4j/src/main/java/io/cloudsoft/winrm4j/winrm/WinRmTool.java

@@ -4,15 +4,17 @@
 import java.nio.charset.Charset;
 import java.util.List;
 import java.util.Map;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import javax.net.ssl.HostnameVerifier;
 
-import io.cloudsoft.winrm4j.client.WinRmClient;
 import org.apache.http.client.config.AuthSchemes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.cloudsoft.winrm4j.client.WinRmClient;
+import io.cloudsoft.winrm4j.client.WinRmClientContext;
 
 /**
  * Tool for executing commands over WinRM.
@@ -35,7 +37,8 @@ public class WinRmTool {
     private final boolean disableCertificateChecks;
     private final String workingDirectory;
     private final Map<String, String> environment;
-    private HostnameVerifier hostnameVerifier;
+    private final HostnameVerifier hostnameVerifier;
+    private final WinRmClientContext context;
 
     public static class Builder {
         private String authenticationScheme = AuthSchemes.NTLM;
@@ -49,6 +52,7 @@ public static class Builder {
         private String workingDirectory;
         private Map<String, String> environment;
         private HostnameVerifier hostnameVerifier;
+        private WinRmClientContext context;
 
         private static final Pattern matchPort = Pattern.compile(".*:(\\d+)$");
 
@@ -99,9 +103,18 @@ public Builder port(int port) {
             this.port = port;
             return this;
         }
+
+        public Builder context(WinRmClientContext context) {
+            this.context = context;
+            return this;
+        }
 
         public WinRmTool build() {
-            return new WinRmTool(getEndpointUrl(address, useHttps, port), domain, username, password, authenticationScheme, disableCertificateChecks, workingDirectory, environment, hostnameVerifier);
+            return new WinRmTool(getEndpointUrl(address, useHttps, port),
+                    domain, username, password, authenticationScheme,
+                    disableCertificateChecks, workingDirectory,
+                    environment, hostnameVerifier,
+                    context);
         }
 
         // TODO remove arguments when method WinRmTool.connect() is removed
@@ -136,12 +149,11 @@ private static String getEndpointUrl(String address, Boolean useHttps, Integer p
         }
     }
 
-    @Deprecated
-    public static WinRmTool connect(String address, String username, String password) {
-        return new WinRmTool(WinRmTool.Builder.getEndpointUrl(address, false, DEFAULT_WINRM_PORT), null, username, password, AuthSchemes.NTLM, false, null, null, null);
-    }
-
-    private WinRmTool(String address, String domain, String username, String password, String authenticationScheme, boolean disableCertificateChecks, String workingDirectory, Map<String, String> environment, HostnameVerifier hostnameVerifier) {
+    private WinRmTool(String address, String domain, String username,
+            String password, String authenticationScheme,
+            boolean disableCertificateChecks, String workingDirectory,
+            Map<String, String> environment, HostnameVerifier hostnameVerifier,
+            WinRmClientContext context) {
         this.disableCertificateChecks = disableCertificateChecks;
         this.address = address;
         this.domain = domain;
@@ -151,6 +163,7 @@ private WinRmTool(String address, String domain, String username, String passwor
         this.workingDirectory = workingDirectory;
         this.environment = environment;
         this.hostnameVerifier = hostnameVerifier;
+        this.context = context;
     }
 
     /**
@@ -217,6 +230,9 @@ public WinRmToolResponse executeCommand(String command) {
         if (retriesForConnectionFailures != null) {
             builder.retriesForConnectionFailures(retriesForConnectionFailures);
         }
+        if (context != null) {
+            builder.context(context);
+        }
 
         WinRmClient client = builder.build();
 
@@ -309,4 +325,5 @@ private String join(List<String> commands, String delim, boolean endWithDelim) {
         }
         return builder.toString();
     }
+
 }

winrm4j/src/test/java/io/cloudsoft/winrm4j/winrm/WinRmToolExecLiveTest.java

@@ -398,6 +398,13 @@ public void testToolReuse() throws Exception {
 
     @Test(groups="Live")
     public void testToolConcurrentReuse() throws Exception {
+        // There are built-in retries at two levels in the code:
+        // * executePs will retry 10 times
+        // * each low-level WinRm command will retry 16 times
+        // * WinRm.delete is called in a finally, again retrying for 16 times
+        // As a result each executePs call could retry requests for a total of
+        // 10 x (16 + 16) = 320 times PER ITERATION
+        // That would result in 3200 failing requests for the tasks below!
         final int NUM_RUNS = 10;
         final int TIMEOUT_MINS = 30;
         final AtomicInteger counter = new AtomicInteger();