Improve Brooklyn-management of OSGi-already-managed bundles

core/src/main/java/org/apache/brooklyn/core/catalog/internal/BasicBrooklynCatalog.java

@@ -1782,7 +1782,8 @@ public void addItem(CatalogItem<?,?> item) {
         //assume forceUpdate for backwards compatibility
         log.debug("Adding manual catalog item to "+mgmt+": "+item);
         checkNotNull(item, "item");
-        CatalogUtils.installLibraries(mgmt, item.getLibraries());
+        //don't activate bundles; only intended for legacy tests where that might not work
+        CatalogUtils.installLibraries(mgmt, item.getLibraries(), false);
         if (manualAdditionsCatalog==null) loadManualAdditionsCatalog();
         manualAdditionsCatalog.addEntry(getAbstractCatalogItem(item));
     }

core/src/main/java/org/apache/brooklyn/core/catalog/internal/CatalogUtils.java

@@ -168,6 +168,11 @@ public static BrooklynClassLoadingContext newClassLoadingContextForCatalogItems(
      * Registers all bundles with the management context's OSGi framework.
      */
     public static void installLibraries(ManagementContext managementContext, @Nullable Collection<CatalogBundle> libraries) {
+        installLibraries(managementContext, libraries, true);
+    }
+    /** As {@link #installLibraries(ManagementContext, Collection)} but letting caller suppress the deferred start/install
+     * (for use in tests where bundles' BOMs aren't resolvable). */
+    public static void installLibraries(ManagementContext managementContext, @Nullable Collection<CatalogBundle> libraries, boolean startBundlesAndInstallToBrooklyn) {
         if (libraries == null) return;
 
         ManagementContextInternal mgmt = (ManagementContextInternal) managementContext;
@@ -190,16 +195,18 @@ public static void installLibraries(ManagementContext managementContext, @Nullab
                 results.add(result);
             }
             Map<String, Throwable> errors = MutableMap.of();
-            for (OsgiBundleInstallationResult r: results) {
-                if (r.getDeferredStart()!=null) {
-                    try {
-                        r.getDeferredStart().run();
-                    } catch (Throwable t) {
-                        Exceptions.propagateIfFatal(t);
-                        // above will done rollback for the failed item, but we need consistent behaviour for all libraries;
-                        // for simplicity we simply have each bundle either fully installed or fully rolled back
-                        // (alternative would be to roll back everything)
-                        errors.put(r.getVersionedName().toString(), t);
+            if (startBundlesAndInstallToBrooklyn) {
+                for (OsgiBundleInstallationResult r: results) {
+                    if (r.getDeferredStart()!=null) {
+                        try {
+                            r.getDeferredStart().run();
+                        } catch (Throwable t) {
+                            Exceptions.propagateIfFatal(t);
+                            // above will done rollback for the failed item, but we need consistent behaviour for all libraries;
+                            // for simplicity we simply have each bundle either fully installed or fully rolled back
+                            // (alternative would be to roll back everything)
+                            errors.put(r.getVersionedName().toString(), t);
+                        }
                     }
                 }
             }
@@ -215,8 +222,8 @@ public static void installLibraries(ManagementContext managementContext, @Nullab
             }
             if (log.isDebugEnabled()) { 
                 logDebugOrTraceIfRebinding(log, 
-                    "Registered {} bundles in {}",
-                    new Object[]{libraries.size(), Time.makeTimeStringRounded(timer)});
+                    "Registered {} bundle{} in {}",
+                    new Object[]{libraries.size(), Strings.s(libraries.size()), Time.makeTimeStringRounded(timer)});
             }
         }
     }

core/src/main/java/org/apache/brooklyn/core/mgmt/ha/OsgiManager.java

@@ -393,6 +393,10 @@ public OsgiBundleInstallationResult uninstallUploadedBundle(ManagedBundle bundle
      * This does not throw but returns a reference containing errors and result for caller to inspect and handle. 
      */
     public ReferenceWithError<OsgiBundleInstallationResult> uninstallUploadedBundle(ManagedBundle bundleMetadata, boolean force) {
+        return uninstallUploadedBundle(bundleMetadata, force, false);
+    }
+
+    public ReferenceWithError<OsgiBundleInstallationResult> uninstallUploadedBundle(ManagedBundle bundleMetadata, boolean force, boolean leaveInOsgi) {
         OsgiBundleInstallationResult result = new OsgiBundleInstallationResult();
         result.metadata = bundleMetadata;
         List<Throwable> errors = MutableList.of();
@@ -425,22 +429,24 @@ public ReferenceWithError<OsgiBundleInstallationResult> uninstallUploadedBundle(
                 errors.add(e);            
             }
 
-            Bundle bundle = framework.getBundleContext().getBundle(bundleMetadata.getOsgiUniqueUrl());
-            result.bundle = bundle;
-            if (bundle==null) {
-                Exception e = new IllegalStateException("No such bundle installed in OSGi when uninstalling: "+bundleMetadata);
-                if (!force) Exceptions.propagate(e);
-                log.warn(e.getMessage());
-                errors.add(e);
-            } else {
-                try {
-                    bundle.stop();
-                    bundle.uninstall();
-                } catch (BundleException e) {
-                    Exceptions.propagateIfFatal(e);
+            if (!leaveInOsgi) {
+                Bundle bundle = framework.getBundleContext().getBundle(bundleMetadata.getOsgiUniqueUrl());
+                result.bundle = bundle;
+                if (bundle==null) {
+                    Exception e = new IllegalStateException("No such bundle installed in OSGi when uninstalling: "+bundleMetadata);
                     if (!force) Exceptions.propagate(e);
-                    log.warn("Error stopping and uninstalling "+bundleMetadata+": "+e);
-                    errors.add(e);            
+                    log.warn(e.getMessage());
+                    errors.add(e);
+                } else {
+                    try {
+                        bundle.stop();
+                        bundle.uninstall();
+                    } catch (BundleException e) {
+                        Exceptions.propagateIfFatal(e);
+                        if (!force) Exceptions.propagate(e);
+                        log.warn("Error stopping and uninstalling "+bundleMetadata+": "+e);
+                        errors.add(e);            
+                    }
                 }
             }
         } catch (Exception e) {

core/src/main/java/org/apache/brooklyn/util/core/ClassLoaderUtils.java

@@ -330,8 +330,11 @@ protected <T> Maybe<T> tryLoadFromBundle(LoaderDispatcher<T> dispatcher, String
             }
             return dispatcher.tryLoadFrom(bundle.get(), name);
         } else {
-            log.warn("Request for bundle '"+symbolicName+"' "+(Strings.isNonBlank(version) ? "("+version+") " : "")+"will be ignored as no framework available; will look for '"+name+"' in plain old classpath");
-            return dispatcher.tryLoadFrom(classLoader, name);
+            Maybe<T> result = dispatcher.tryLoadFrom(classLoader, name);
+            if (result.isAbsent()) {
+                log.warn("Request for bundle '"+symbolicName+"' "+(Strings.isNonBlank(version) ? "("+version+") " : "")+"was ignored as no framework available; and failed to find '"+name+"' in plain old classpath");
+            }
+            return result;
         }
     }
 

core/src/main/java/org/apache/brooklyn/util/core/osgi/BundleMaker.java

@@ -23,6 +23,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
+import java.net.URL;
 import java.util.Enumeration;
 import java.util.Map;
 import java.util.Set;
@@ -298,16 +299,29 @@ private boolean addUrlItemRecursively(ZipOutputStream zout, String root, String
             return false;
         }
 
-        if (!addUrlDirToZipRecursively(zout, root, item, itemFound, filter)) {
-            // can't reliably tell if item a file or a folder (listing files), esp w classpath where folder is just a list of files, 
-            // so try the latter and see if it succeeds
-            // slightly inefficient but should work fine
-            // only issue is that an empty dir and a file of size 0 will appear identical?
+        if (isKnownNotToBeADirectoryListing(item) || !
+            // can't reliably tell if item a file or a folder (listing files), esp w classpath where folder is treated as a list of files, 
+            // so if we can't tell try it as a list of files; not guaranteed, and empty dir and a file of size 0 will appear identical, but better than was
+            // (mainly used for tests)
+                addUrlDirToZipRecursively(zout, root, item, itemFound, filter)) {
             addUrlFileToZip(zout, root, item, filter);
         }
         return true;
     }
 
+    private boolean isKnownNotToBeADirectoryListing(String item) {
+        try { 
+            URL url = new URL(item);
+            if (url.getProtocol().equals("file")) {
+                return !new File(url.getFile()).isDirectory();
+            }
+        } catch (Exception e) {
+            Exceptions.propagateIfFatal(e);
+            // ignore otherwise -- probably unknown protocol
+        }
+        return false;
+    }
+
     private boolean addUrlDirToZipRecursively(ZipOutputStream zout, String root, String item, InputStream itemFound, Predicate<? super String> filter) throws IOException {
         LineReader lr = new LineReader(new InputStreamReader(itemFound));
         boolean readSubdirFile = false;
@@ -325,7 +339,8 @@ private boolean addUrlDirToZipRecursively(ZipOutputStream zout, String root, Str
                     // not a folder
                     return false;
                 } else {
-                    // previous entry made clear it was a folder, but this one didn't work!
+                    // previous entry suggested it was a folder, but this one didn't work! -- was a false positive
+                    // but zip will be in inconsistent state, so throw
                     throw new IllegalStateException("Failed to read entry "+line+" in "+item+" but previous entry implied it was a directory");
                 }
             }
@@ -364,7 +379,7 @@ public File createTempBundle(String nameHint, Manifest mf, Map<ZipEntry, InputSt
             zout = mf!=null ? new JarOutputStream(new FileOutputStream(f2), mf) : new ZipOutputStream(new FileOutputStream(f2));
             writeZipEntries(zout, files);
         } catch (IOException e) {
-            throw Exceptions.propagate("Unable to read/write for "+nameHint, e);
+            throw Exceptions.propagateAnnotated("Unable to read/write for "+nameHint, e);
         } finally {
             Streams.closeQuietly(zf);
             Streams.closeQuietly(zout);

core/src/test/java/org/apache/brooklyn/util/core/ClassLoaderUtilsTest.java

@@ -58,7 +58,6 @@
 import org.apache.brooklyn.util.maven.MavenArtifact;
 import org.apache.brooklyn.util.maven.MavenRetriever;
 import org.apache.brooklyn.util.osgi.OsgiTestResources;
-import org.apache.brooklyn.util.osgi.OsgiUtils;
 import org.apache.commons.io.output.ByteArrayOutputStream;
 import org.osgi.framework.Bundle;
 import org.osgi.framework.launch.Framework;
@@ -216,7 +215,10 @@ public void testLoadClassInOsgiCore() throws Exception {
 
         mgmt = LocalManagementContextForTests.builder(true).enableOsgiReusable().build();
         Bundle bundle = getBundle(mgmt, "org.apache.brooklyn.core");
-        Entity entity = createSimpleEntity(bundle.getLocation(), clazz);
+        String url = bundle.getLocation();
+        // NB: the above will be a system:file: url when running tests against target/classes/ -- but
+        // OSGi manager will accept that if running in dev mode
+        Entity entity = createSimpleEntity(url, clazz);
 
         ClassLoaderUtils cluMgmt = new ClassLoaderUtils(getClass(), mgmt);
         ClassLoaderUtils cluClass = new ClassLoaderUtils(clazz);

utils/common/src/main/java/org/apache/brooklyn/util/stream/Streams.java

@@ -29,6 +29,7 @@
 import java.io.Reader;
 import java.io.StringReader;
 import java.io.Writer;
+import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.security.DigestInputStream;
 import java.security.MessageDigest;
@@ -104,6 +105,18 @@ public static byte[] readFully(InputStream is) {
         }
     }
 
+    /** reads the input stream fully into the given byte buffer or until the supplied buffer is full,
+     * returning the number of bytes read */
+    public static int readFully(InputStream s, byte[] buf) throws IOException {
+        int count = 0;
+        while (count < buf.length) {
+            int countHere = s.read(buf, count, buf.length-count);
+            if (countHere<=0) return count;
+            count += countHere;
+        }
+        return count;
+    }
+
     public static byte[] readFullyAndClose(InputStream is) {
         try {
             return readFully(is);
@@ -234,5 +247,29 @@ public static String getMd5Checksum(InputStream in) {
         }
         return result.toString().toUpperCase();
     }
-
+
+    /** True iff the input streams read to completion give identical contents. Streams are closed afterwards. */
+    public static boolean compare(InputStream s1, InputStream s2) throws IOException {
+        try {
+            if (s1==null || s2==null) {
+                return s1==null && s2==null;
+            }
+            byte[] buf1 = new byte[4096], buf2 = new byte[4096];
+            while (true) {
+                int r1 = readFully(s1, buf1);
+                int r2 = readFully(s2, buf2);
+                // do this in case HeapByteBuffer.equals has an efficient intrinsic comparison;
+                // curiously there is no efficient array-compare a la System.arraycopy;
+                // the only thing I've found is that Arrays.equals(char[], char[]) is optimized/intrinsic,
+                // and allegedly 8 times faster than all other array comparison routines!
+                // https://stackoverflow.com/questions/41153992/why-is-arrays-equalschar-char-8-times-faster-than-all-the-other-versions
+                if (!ByteBuffer.wrap(buf1,0,r1).equals(ByteBuffer.wrap(buf2,0,r2))) return false;
+                if (r1<4096) return true;
+            }
+        } finally {
+            closeQuietly(s1);
+            closeQuietly(s2);
+        }
+    }
+
 }

utils/common/src/test/java/org/apache/brooklyn/util/stream/StreamsTest.java

@@ -19,7 +19,9 @@
 package org.apache.brooklyn.util.stream;
 
 import java.io.ByteArrayInputStream;
+import java.io.IOException;
 
+import org.apache.brooklyn.util.text.Strings;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
@@ -33,4 +35,19 @@ public void testChecksum() {
             );
 
     }
+
+    @Test
+    public void testComparison() throws IOException {
+        Assert.assertTrue(Streams.compare(Streams.newInputStreamWithContents("hello"), Streams.newInputStreamWithContents("hello")));
+        Assert.assertFalse(Streams.compare(Streams.newInputStreamWithContents("hello"), Streams.newInputStreamWithContents("hello2")));
+        Assert.assertFalse(Streams.compare(Streams.newInputStreamWithContents("hello2"), Streams.newInputStreamWithContents("hello")));
+        Assert.assertFalse(Streams.compare(null, Streams.newInputStreamWithContents("hello")));
+        Assert.assertTrue(Streams.compare(null, null));
+        // test across byte buffers
+        Assert.assertTrue(Streams.compare(Streams.newInputStreamWithContents(Strings.repeat("hello", 1000)), 
+            Streams.newInputStreamWithContents(Strings.repeat("hello", 1000))));
+        Assert.assertFalse(Streams.compare(Streams.newInputStreamWithContents(Strings.repeat("hello", 1000)+"1"), 
+            Streams.newInputStreamWithContents(Strings.repeat("hello", 1000)+"2")));
+    }
+
 }