| /* |
| * 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 org.apache.sling.distribution.journal.it; |
| |
| |
| import static java.lang.Boolean.getBoolean; |
| import static java.lang.String.format; |
| import static java.util.Collections.singletonList; |
| import static java.util.Collections.singletonMap; |
| import static org.apache.kafka.clients.admin.AdminClient.create; |
| import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG; |
| import static org.apache.sling.testing.paxexam.SlingOptions.paxTinybundles; |
| import static org.apache.sling.testing.paxexam.SlingOptions.slingDistribution; |
| import static org.apache.sling.testing.paxexam.SlingOptions.slingQuickstartOak; |
| import static org.ops4j.pax.exam.Constants.START_LEVEL_SYSTEM_BUNDLES; |
| import static org.ops4j.pax.exam.CoreOptions.bootDelegationPackage; |
| import static org.ops4j.pax.exam.CoreOptions.bundle; |
| import static org.ops4j.pax.exam.CoreOptions.composite; |
| import static org.ops4j.pax.exam.CoreOptions.frameworkStartLevel; |
| import static org.ops4j.pax.exam.CoreOptions.keepCaches; |
| import static org.ops4j.pax.exam.CoreOptions.mavenBundle; |
| import static org.ops4j.pax.exam.CoreOptions.repository; |
| import static org.ops4j.pax.exam.CoreOptions.systemProperty; |
| import static org.ops4j.pax.exam.CoreOptions.url; |
| import static org.ops4j.pax.exam.CoreOptions.vmOption; |
| import static org.ops4j.pax.exam.CoreOptions.when; |
| import static org.ops4j.pax.exam.cm.ConfigurationAdminOptions.factoryConfiguration; |
| import static org.ops4j.pax.exam.cm.ConfigurationAdminOptions.newConfiguration; |
| import static org.osgi.util.converter.Converters.standardConverter; |
| |
| import java.io.File; |
| import java.util.Map; |
| |
| import javax.inject.Inject; |
| |
| import org.apache.kafka.clients.admin.AdminClient; |
| import org.apache.kafka.clients.admin.CreateTopicsResult; |
| import org.apache.kafka.clients.admin.NewTopic; |
| import org.apache.sling.distribution.journal.MessagingProvider; |
| import org.apache.sling.distribution.journal.kafka.KafkaClientProvider; |
| import org.apache.sling.distribution.journal.kafka.KafkaEndpoint; |
| import org.apache.sling.distribution.serialization.impl.vlt.VaultDistributionPackageBuilderFactory; |
| import org.apache.sling.testing.paxexam.SlingOptions; |
| import org.apache.sling.testing.paxexam.TestSupport; |
| import org.ops4j.pax.exam.ConfigurationManager; |
| import org.ops4j.pax.exam.Constants; |
| import org.ops4j.pax.exam.CoreOptions; |
| import org.ops4j.pax.exam.Option; |
| import org.ops4j.pax.exam.options.CompositeOption; |
| import org.ops4j.pax.exam.options.DefaultCompositeOption; |
| import org.ops4j.pax.exam.options.libraries.JUnitBundlesOption; |
| import org.osgi.framework.BundleContext; |
| import org.osgi.service.cm.ConfigurationAdmin; |
| |
| import com.google.common.collect.ImmutableMap; |
| |
| public class DistributionTestSupport extends TestSupport { |
| public static final String TOPIC_PACKAGE = "aemdistribution_package"; |
| public static final String TOPIC_DISCOVERY = "aemdistribution_discovery"; |
| public static final String TOPIC_COMMAND = "aemdistribution_command"; |
| public static final String TOPIC_STATUS = "aemdistribution_status"; |
| public static final String TOPIC_EVENT = "aemdistribution_event"; |
| |
| @Inject |
| protected BundleContext bundleContext; |
| |
| @Inject |
| protected ConfigurationAdmin configAdmin; |
| |
| private int httpPort = 8181; |
| |
| public DistributionTestSupport withHttpPort(int httpPort) { |
| this.httpPort = httpPort; |
| return this; |
| } |
| |
| public Option baseConfiguration() { |
| String workingDirectory = workingDirectory(); |
| FileUtil.deleteDir(new File(workingDirectory)); |
| return baseConfiguration(workingDirectory); |
| } |
| |
| public Option baseConfiguration(String baseDirectory) { |
| // Patch versions of features provided by SlingOptions |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.commons.mime"); |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.commons.metrics"); |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.distribution.core"); |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.distribution.journal"); |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.distribution.journal.messages"); |
| SlingOptions.versionResolver.setVersionFromProject("org.apache.sling", "org.apache.sling.distribution.journal.kafka"); |
| SlingOptions.versionResolver.setVersionFromProject("io.dropwizard.metrics", "metrics-core"); |
| SlingOptions.versionResolver.setVersion("org.slf4j", "log4j-over-slf4j", "1.7.6"); |
| |
| Option baseOptions = composite( |
| failOnUnresolvedBundles(), |
| keepCaches(), |
| localMavenRepo(), |
| //repository("https://repository.apache.org/snapshots/").id("apache-snapshots").allowSnapshots(), |
| repository("https://repo1.maven.org/maven2").id("central"), |
| CoreOptions.workingDirectory(workingDirectory()), |
| mavenBundle().groupId("org.apache.sling").artifactId("org.apache.sling.testing.paxexam").versionAsInProject(), |
| paxTinybundles(), |
| SlingOptions.logback(), |
| mavenBundle().groupId("org.slf4j").artifactId("log4j-over-slf4j").version(SlingOptions.versionResolver), |
| mvn("com.google.code.gson", "gson"), |
| |
| // The base sling Quickstart |
| slingQuickstart(baseDirectory), |
| mavenBundle().groupId("org.apache.felix").artifactId("org.apache.felix.webconsole.plugins.ds").version(SlingOptions.versionResolver), |
| mavenBundle().groupId("org.apache.sling").artifactId("org.apache.sling.commons.metrics").version(SlingOptions.versionResolver), |
| mvn("org.apache.felix", "org.apache.felix.rootcause"), |
| mvn("org.apache.felix", "org.apache.felix.systemready"), |
| kafka(), |
| |
| // The bundle built (org.apache.sling.distribution.journal) |
| mvn("org.apache.sling", "org.apache.sling.distribution.journal"), |
| mvn("org.apache.sling", "org.apache.sling.distribution.journal.messages"), |
| mvn("org.apache.sling", "org.apache.sling.distribution.journal.kafka"), |
| |
| // distribution bundles |
| slingDistribution(), |
| |
| // testing |
| //slingResourcePresence(), // see https://github.com/apache/sling-org-apache-sling-resource-presence |
| jsoup(), |
| myJunitBundles() |
| |
| ); |
| |
| // Remote debugger on the forked JVM |
| // Run with mvn install -DisDebugEnabled=true |
| return getBoolean("isDebugEnabled") |
| ? composite(remoteDebug(), baseOptions) |
| : baseOptions; |
| } |
| |
| public static CompositeOption myJunitBundles() { |
| return new DefaultCompositeOption( |
| composite(defaultTestSystemOptions()), |
| new JUnitBundlesOption(), |
| systemProperty("pax.exam.invoker").value("junit"), |
| mvn("org.mockito", "mockito-all"), |
| mvn("org.apache.servicemix.bundles", "org.apache.servicemix.bundles.hamcrest"), |
| mvn("org.awaitility", "awaitility"), |
| bundle("link:classpath:META-INF/links/org.ops4j.pax.exam.invoker.junit.link")); |
| } |
| |
| /** |
| * Dependencies for mockito 2 |
| */ |
| public static Option mockito2() { |
| return composite( |
| mvn("org.objenesis", "objenesis"), |
| mvn("net.bytebuddy", "byte-buddy"), |
| mvn("net.bytebuddy", "byte-buddy-agent"), |
| mvn("org.mockito", "mockito-core") |
| ); |
| } |
| |
| /** |
| * Standard test system options with just at inject removed as it collides with |
| * the one provided in sling |
| */ |
| private static Option[] defaultTestSystemOptions() { |
| ConfigurationManager cm = new ConfigurationManager(); |
| String logging = cm.getProperty(Constants.EXAM_LOGGING_KEY, |
| Constants.EXAM_LOGGING_PAX_LOGGING); |
| |
| return new Option[] { |
| bootDelegationPackage("sun.*"), |
| frameworkStartLevel(Constants.START_LEVEL_TEST_BUNDLE), |
| url("link:classpath:META-INF/links/org.ops4j.pax.exam.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.exam.inject.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.extender.service.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.osgi.compendium.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| |
| when(logging.equals(Constants.EXAM_LOGGING_PAX_LOGGING)).useOptions( |
| url("link:classpath:META-INF/links/org.ops4j.pax.logging.api.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES)), |
| |
| url("link:classpath:META-INF/links/org.ops4j.base.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.swissbox.core.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.swissbox.extender.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.swissbox.framework.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.swissbox.lifecycle.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| url("link:classpath:META-INF/links/org.ops4j.pax.swissbox.tracker.link").startLevel( |
| START_LEVEL_SYSTEM_BUNDLES), |
| /* |
| url("link:classpath:META-INF/links/org.apache.geronimo.specs.atinject.link") |
| .startLevel(START_LEVEL_SYSTEM_BUNDLES) }; |
| */ |
| }; |
| } |
| |
| protected Option slingQuickstart(String baseDirectory) { |
| final String workingDirectory = String.format("%s/instance", baseDirectory); |
| final String dataStoreDirectory = String.format("%s/shareddatastore", baseDirectory); |
| System.out.println(String.format("Quickstart with workingDirectory %s, port %s", workingDirectory, httpPort)); |
| return slingQuickstartOakSharedBlobstore(workingDirectory, dataStoreDirectory, httpPort); |
| } |
| |
| public static Option defaultOsgiConfigs() { |
| return defaultOsgiConfigs(""); |
| } |
| |
| /** |
| * OSGI configurations targeted to author and publish instances |
| */ |
| protected static Option defaultOsgiConfigs(String journalEndpoint) { |
| return composite( |
| newConfiguration("org.apache.sling.jcr.resource.internal.JcrSystemUserValidator") |
| .put("allow.only.system.user", false).asOption(), |
| |
| // For production the users would be: replication-service,content-writer-service |
| factoryConfiguration("org.apache.sling.serviceusermapping.impl.ServiceUserMapperImpl.amended") |
| .put("user.mapping", new String[]{"org.apache.sling.distribution.journal:bookkeeper=admin","org.apache.sling.distribution.journal:importer=admin"}) |
| .asOption(), |
| |
| factoryConfiguration(VaultDistributionPackageBuilderFactory.class.getName()) |
| .put("name", "journal") |
| .put("type", "inmemory") |
| .put("useBinaryReferences", "true") |
| .put("aclHandling", "IGNORE") |
| .put("package.filters", new String[]{"/home/users|-.*/.tokens", "/home/users|-.*/rep:cache"}) |
| .put("property.filters", new String[]{"/|-^.*/cq:lastReplicated|-^.*/cq:lastReplicatedBy|-^.*/cq:lastReplicationAction"}) |
| .asOption(), |
| |
| newConfiguration("org.apache.sling.distribution.journal.kafka.KafkaClientProvider") |
| .asOption(), |
| |
| newConfiguration("org.apache.sling.distribution.component.impl.DistributionComponentFactoryMap") |
| .put("mapping.agent", new String[]{// |
| "pub:org.apache.sling.distribution.journal.impl.publisher.DistributionPublisherFactory"}) |
| .asOption(), |
| |
| newConfiguration("org.apache.sling.distribution.journal.impl.shared.JournalAvailableChecker") |
| .put("scheduler.period", 1L) |
| .asOption() |
| |
| ); |
| |
| } |
| |
| /** |
| * OSGI configurations targeted to the author instances only |
| */ |
| public static Option authorOsgiConfigs() { |
| return composite( |
| factoryConfiguration("org.apache.sling.distribution.journal.impl.publisher.DistributionPublisherFactory") |
| .put("name", "agent1") |
| .put("packageBuilder.target", "(name=journal)") |
| .asOption(), |
| factoryConfiguration("org.apache.sling.distribution.resources.impl.DistributionServiceResourceProviderFactory") |
| .put("kind", "agent") |
| .put("provider.roots", "/libs/sling/distribution/services/agents") |
| .asOption(), |
| factoryConfiguration("org.apache.sling.distribution.resources.impl.DistributionServiceResourceProviderFactory") |
| .put("kind", "exporter") |
| .put("provider.roots", "/libs/sling/distribution/services/exporters") |
| .asOption(), |
| factoryConfiguration("org.apache.sling.distribution.resources.impl.DistributionServiceResourceProviderFactory") |
| .put("kind", "importer") |
| .put("provider.roots", "/libs/sling/distribution/services/importers") |
| .asOption() |
| ); |
| |
| } |
| |
| protected static Option publishOsgiConfigs() { |
| return publishOsgiConfigs("subscriber-agent1"); |
| } |
| |
| /** |
| * OSGI configuration targeted to the publish instances only |
| */ |
| public static Option publishOsgiConfigs(String agentName) { |
| return publishOsgiConfigs(agentName, true, false); |
| } |
| |
| protected static Option publishOsgiConfigs(String agentName, boolean editable, boolean stagingPrecondition) { |
| return composite( |
| factoryConfiguration("org.apache.sling.distribution.resources.impl.DistributionServiceResourceProviderFactory") |
| .put("kind", "agent") |
| .put("provider.roots", "/libs/sling/distribution/services/agents") |
| .asOption(), |
| |
| factoryConfiguration("org.apache.sling.distribution.journal.impl.subscriber.DistributionSubscriberFactory") |
| .put("name", agentName) |
| .put("agentNames", new String[]{ "agent1"}) |
| .put("packageBuilder.target", "(name=journal)") |
| .put("precondition.target", stagingPrecondition ? "(name=staging)" : "(name=default)") |
| .put("editable", editable) |
| .put("announceDelay", "500") |
| .asOption()); |
| } |
| |
| |
| protected static Option remoteDebug() { |
| return remoteDebug(5005); |
| } |
| |
| protected static Option remoteDebug(int debugPort) { |
| System.out.println(String.format("Remote debugger on port: %s", debugPort)); |
| return vmOption("-Xrunjdwp:transport=dt_socket,server=y,suspend=n,address=5005"); |
| } |
| |
| private static Option slingQuickstartOakSharedBlobstore(String workingDirectory, String dataStoreDirectory, int httpPort) { |
| |
| // Option for a quickstart running an Oak repository with a nodestore and a shared data store. |
| |
| String slingHome = String.format("%s/sling", workingDirectory); |
| String repositoryHome = String.format("%s/repository", slingHome); |
| String localIndexDir = String.format("%s/index", repositoryHome); |
| return composite( |
| |
| slingQuickstartOak(), |
| |
| mavenBundle() |
| .groupId("org.apache.jackrabbit") |
| .artifactId("oak-lucene") |
| .version(SlingOptions.versionResolver), |
| |
| mavenBundle() |
| .groupId("org.apache.jackrabbit") |
| .artifactId("oak-segment-tar") |
| .version(SlingOptions.versionResolver), |
| |
| mavenBundle() |
| .groupId("org.apache.sling") |
| .artifactId("org.apache.sling.jcr.oak.server") |
| .version(SlingOptions.versionResolver), |
| |
| newConfiguration("org.apache.felix.http") |
| .put("org.osgi.service.http.port", httpPort).asOption(), |
| |
| newConfiguration("org.apache.jackrabbit.oak.segment.SegmentNodeStoreService") |
| .put("customBlobStore", true) |
| .put("repository.home", repositoryHome) |
| .put("name", "NodeStore with custom blob store").asOption(), |
| |
| newConfiguration("org.apache.jackrabbit.oak.plugins.blob.datastore.FileDataStore") |
| .put("path", dataStoreDirectory) |
| .put("minRecordLength", 16384).asOption(), |
| |
| newConfiguration("org.apache.jackrabbit.oak.plugins.index.lucene.LuceneIndexProviderService") |
| .put("localIndexDir", localIndexDir).asOption() |
| ); |
| } |
| |
| private static Option kafka() { |
| return composite( |
| mvn("com.fasterxml.jackson.core", "jackson-core"), |
| mvn("com.fasterxml.jackson.core", "jackson-annotations"), |
| mvn("com.fasterxml.jackson.core", "jackson-databind"), |
| mvn("org.apache.servicemix.bundles", "org.apache.servicemix.bundles.kafka-clients") |
| ); |
| } |
| |
| public static Option mvn(String groupId, String artifactId) { |
| return mavenBundle().groupId(groupId).artifactId(artifactId).versionAsInProject(); |
| } |
| |
| private static Option jsoup() { |
| return mavenBundle().groupId("org.jsoup").artifactId("jsoup").versionAsInProject(); |
| } |
| |
| public static void createTopic(String topicName) { |
| NewTopic newTopic = new NewTopic(topicName, 1, (short) 1); |
| try (AdminClient admin = create(singletonMap(BOOTSTRAP_SERVERS_CONFIG, |
| standardConverter().convert(kafkaProperties()).to(KafkaEndpoint.class).kafkaBootstrapServers()))) { |
| CreateTopicsResult result = admin.createTopics(singletonList(newTopic)); |
| result.values().get(topicName).get(); |
| } catch (Exception e) { |
| throw new RuntimeException(format("Failed to create topic %s", topicName), e); |
| } |
| } |
| |
| public static void createTopics() { |
| createTopic(TOPIC_DISCOVERY); |
| createTopic(TOPIC_PACKAGE); |
| createTopic(TOPIC_COMMAND); |
| createTopic(TOPIC_STATUS); |
| createTopic(TOPIC_EVENT); |
| } |
| |
| public static MessagingProvider createProvider() { |
| KafkaClientProvider provider = new KafkaClientProvider(); |
| provider.activate(standardConverter().convert(kafkaProperties()).to(KafkaEndpoint.class)); |
| return provider; |
| } |
| |
| private static Map<String,String> kafkaProperties() { |
| return ImmutableMap.of( |
| "kafkaDefaultApiTimeout", "5000", |
| "kafkaConnectTimeout", "32000"); |
| } |
| |
| } |
