src/main/java/org/apache/sling/distribution/journal/impl/publisher/DistributionPublisher.java - sling-org-apache-sling-distribution-journal - Git at Google

 /*
  * 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.impl.publisher;


 import static org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService.timed;
 import static java.util.Objects.requireNonNull;
 import static org.apache.sling.distribution.DistributionRequestState.ACCEPTED;
 import static org.apache.sling.distribution.DistributionRequestType.ADD;
 import static org.apache.sling.distribution.DistributionRequestType.DELETE;
 import static org.apache.sling.distribution.DistributionRequestType.TEST;

 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Dictionary;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Hashtable;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;

 import javax.annotation.CheckForNull;
 import javax.annotation.Nonnull;
 import javax.annotation.ParametersAreNonnullByDefault;
 import javax.management.NotCompliantMBeanException;

 import org.apache.sling.distribution.journal.impl.event.DistributionEvent;
 import org.apache.sling.distribution.journal.impl.queue.PubQueueProvider;
 import org.apache.sling.distribution.journal.impl.shared.AgentState;
 import org.apache.sling.distribution.journal.impl.shared.DefaultDistributionLog;
 import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
 import org.apache.sling.distribution.journal.impl.shared.SimpleDistributionResponse;
 import org.apache.sling.distribution.journal.impl.shared.Topics;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.sling.api.resource.ResourceResolver;
 import org.apache.sling.distribution.DistributionRequest;
 import org.apache.sling.distribution.DistributionRequestState;
 import org.apache.sling.distribution.DistributionRequestType;
 import org.apache.sling.distribution.DistributionResponse;
 import org.apache.sling.distribution.agent.DistributionAgentState;
 import org.apache.sling.distribution.agent.spi.DistributionAgent;
 import org.apache.sling.distribution.common.DistributionException;
 import org.apache.sling.distribution.log.spi.DistributionLog;
 import org.apache.sling.distribution.packaging.DistributionPackageBuilder;
 import org.apache.sling.distribution.queue.spi.DistributionQueue;
 import org.osgi.framework.BundleContext;
 import org.osgi.framework.ServiceRegistration;
 import org.osgi.service.component.annotations.Activate;
 import org.osgi.service.component.annotations.Component;
 import org.osgi.service.component.annotations.Deactivate;
 import org.osgi.service.component.annotations.Reference;
 import org.osgi.service.event.Event;
 import org.osgi.service.event.EventAdmin;
 import org.osgi.service.metatype.annotations.Designate;

 import org.apache.sling.distribution.journal.impl.shared.JMXRegistration;
 import org.apache.sling.distribution.journal.messages.Messages.PackageMessage;
 import org.apache.sling.distribution.journal.MessageSender;
 import org.apache.sling.distribution.journal.MessagingProvider;
 import org.apache.sling.distribution.journal.JournalAvailable;

 /**
  * A Publisher SCD agent which produces messages to be consumed by a {@code DistributionSubscriber} agent.
  */
 @Component(
         service = {},
         immediate = true,
         configurationPid = "org.apache.sling.distribution.journal.impl.publisher.DistributionPublisherFactory"
 )
 @Designate(ocd = PublisherConfiguration.class, factory = true)
 @ParametersAreNonnullByDefault
 public class DistributionPublisher implements DistributionAgent {

     private final Map<DistributionRequestType, Consumer<PackageMessage>> REQ_TYPES = new HashMap<>();

     private final DefaultDistributionLog log;

     @Reference
     private MessagingProvider messagingProvider;

     @Reference(name = "packageBuilder")
     private DistributionPackageBuilder packageBuilder;

     @Reference
     private PackageQueuedNotifier queuedNotifier;

     @Reference
     private PubQueueProvider pubQueueProvider;

     @Reference
     private DiscoveryService discoveryService;

     @Reference
     private PackageMessageFactory factory;

     @Reference
     private EventAdmin eventAdmin;

     @Reference
     private Topics topics;

     @Reference
     JournalAvailable journalAvailable;

     @Reference
     private DistributionMetricsService distributionMetricsService;

     private String pubAgentName;

     private String pkgType;

     private long queuedTimeout;

     private ServiceRegistration<DistributionAgent> componentReg;

     private MessageSender<PackageMessage> sender;

     private JMXRegistration reg;

     public DistributionPublisher() {
         log = new DefaultDistributionLog(pubAgentName, this.getClass(), DefaultDistributionLog.LogLevel.INFO);
         REQ_TYPES.put(ADD,    this::sendAndWait);
         REQ_TYPES.put(DELETE, this::sendAndWait);
         REQ_TYPES.put(TEST,   this::send);
     }

     @Activate
     public void activate(PublisherConfiguration config, BundleContext context) {
         requireNonNull(factory);
         pubAgentName = requireNonNull(config.name());

         queuedTimeout = config.queuedTimeout();

         pkgType = packageBuilder.getType();

         this.sender = messagingProvider.createSender();

         Dictionary<String, Object> props = createServiceProps(config);
         componentReg = requireNonNull(context.registerService(DistributionAgent.class, this, props));

         DistPublisherJMX bean;
         try {
             bean = new DistPublisherJMX(pubAgentName, discoveryService, this);
         } catch (NotCompliantMBeanException e) {
             throw new RuntimeException(e);
         }
         reg = new JMXRegistration(bean, "agent", pubAgentName);

         String msg = String.format("Started Publisher agent %s with packageBuilder %s, queuedTimeout %s",
                 pubAgentName, pkgType, queuedTimeout);
         log.info(msg);
     }

     @Deactivate
     public void deactivate() {
         reg.close();
         componentReg.unregister();
         String msg = String.format("Stopped Publisher agent %s with packageBuilder %s, queuedTimeout %s",
                 pubAgentName, pkgType, queuedTimeout);
         log.info(msg);
     }

     private Dictionary<String, Object> createServiceProps(PublisherConfiguration config) {
         Dictionary<String, Object> props = new Hashtable<>();
         props.put("name", config.name());
         props.put("title", config.name());
         props.put("details", config.name());
         props.put("packageBuilder.target", config.packageBuilder_target());
         props.put("webconsole.configurationFactory.nameHint", config.webconsole_configurationFactory_nameHint());
         return props;
     }

     /**
      * Get queue names for alive subscribed subscriber agents.
      */
     @Nonnull
     @Override
     public Iterable<String> getQueueNames() {

         // Queues names are generated only for the subscriber agents which are
         // alive and are subscribed to the publisher agent name (pubAgentName).
         // The queue names match the subscriber agent identifier (subAgentId).
         //
         // If errors queues are enabled, an error queue name is generated which
         // follows the pattern "%s-error". The pattern is deliberately different
         // from the SCD on Jobs one ("error-%s") as we don't want to support
         // the UI ability to retry items from the error queue.
         Set<String> queueNames = new HashSet<>();
         TopologyView view =  discoveryService.getTopologyView();
         for (String subAgentId : view.getSubscribedAgentIds(pubAgentName)) {
             queueNames.add(subAgentId);
             State subState = view.getState(subAgentId, pubAgentName);
             if (subState != null) {
                 boolean errorQueueEnabled = (subState.getMaxRetries() >= 0);
                 if (errorQueueEnabled) {
                     queueNames.add(String.format("%s-error", subAgentId));
                 }
             }
         }
         return Collections.unmodifiableCollection(queueNames);
     }

     @Override
     public DistributionQueue getQueue(String queueName) {

         // We expect queueName to come from #getQueueNames
         // and this, queueName matches <subAgentId>-error or <subAgentId>
         return queueName.endsWith("-error") ? getErrorQueue(queueName) : getPubQueue(queueName);
     }

     @Nonnull
     private DistributionQueue getErrorQueue(String queueName) {
         AgentId subAgentId = new AgentId(StringUtils.substringBeforeLast(queueName, "-error"));
         return pubQueueProvider.getErrorQueue(pubAgentName, subAgentId.getSlingId(), subAgentId.getAgentName(), queueName);
     }

     @CheckForNull
     private DistributionQueue getPubQueue(String queueName) {
         TopologyView view = discoveryService.getTopologyView();
         AgentId subAgentId = new AgentId(queueName);
         State state = view.getState(subAgentId.getAgentId(), pubAgentName);
         if (state != null) {
             return pubQueueProvider.getQueue(pubAgentName, subAgentId.getSlingId(), subAgentId.getAgentName(), queueName, state.getOffset() + 1, state.getRetries(), state.isEditable());
         }
         return null;
     }


     @Nonnull
     @Override
     public DistributionLog getLog() {
         return log;
     }

     @Nonnull
     @Override
     public DistributionAgentState getState() {
         return AgentState.getState(this);
     }

     @Nonnull
     @Override
     public DistributionResponse execute(ResourceResolver resourceResolver,
                                         DistributionRequest request)
             throws DistributionException {
         Consumer<PackageMessage> handler = REQ_TYPES.get(request.getRequestType());
         if (handler != null) {
             return execute(resourceResolver, request, handler);
         } else {
             return executeUnsupported(request);
         }
     }

     private DistributionResponse execute(ResourceResolver resourceResolver,
                                          DistributionRequest request,
                                          Consumer<PackageMessage> sender)
             throws DistributionException {
         try {
             PackageMessage pkg = timed(distributionMetricsService.getBuildPackageDuration(),
                     () -> factory.create(packageBuilder, resourceResolver, pubAgentName, request)
             );
             timed(distributionMetricsService.getEnqueuePackageDuration(),
                     () -> sender.accept(pkg)
             );
             distributionMetricsService.getExportedPackageSize().update(pkg.getPkgLength());
             distributionMetricsService.getAcceptedRequests().mark();
             String msg = String.format("Distribution request accepted with type %s paths %s ", request.getRequestType(), Arrays.toString(request.getPaths()));
             log.info(msg);
             return new SimpleDistributionResponse(ACCEPTED, msg);
         } catch (Throwable e) {
             distributionMetricsService.getDroppedRequests().mark();
             String msg = String.format("Failed to queue distribution request %s", e.getMessage());
             log.error(msg, e);
             if (e instanceof Error) {
                 throw (Error) e;
             } else {
                 throw new DistributionException(msg, e);
             }
         }
     }

     private void sendAndWait(PackageMessage pkg) {
         try {
             CompletableFuture<Void> received = queuedNotifier.registerWait(pkg.getPkgId());
             Event createdEvent = DistributionEvent.eventPackageCreated(pkg, pubAgentName);
             eventAdmin.postEvent(createdEvent);
             send(pkg);
             received.get(queuedTimeout, TimeUnit.MILLISECONDS);
         } catch (Exception e) {
             queuedNotifier.unRegisterWait(pkg.getPkgId());
             throw new RuntimeException(e);
         }
     }

     private void send(PackageMessage pipePackage) {
         final String topicName = topics.getPackageTopic();
         sender.send(topicName, pipePackage);
     }

     @Nonnull
     private DistributionResponse executeUnsupported(DistributionRequest request) {
         String msg = String.format("Request type %s is not supported by this agent, expected one of %s",
                 request.getRequestType(), REQ_TYPES.keySet());
         log.info(msg);
         return new SimpleDistributionResponse(DistributionRequestState.DROPPED, msg);
     }
 }
	/*
	* 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.impl.publisher;


	import static org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService.timed;
	import static java.util.Objects.requireNonNull;
	import static org.apache.sling.distribution.DistributionRequestState.ACCEPTED;
	import static org.apache.sling.distribution.DistributionRequestType.ADD;
	import static org.apache.sling.distribution.DistributionRequestType.DELETE;
	import static org.apache.sling.distribution.DistributionRequestType.TEST;

	import java.util.Arrays;
	import java.util.Collections;
	import java.util.Dictionary;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.Hashtable;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.CompletableFuture;
	import java.util.concurrent.TimeUnit;
	import java.util.function.Consumer;

	import javax.annotation.CheckForNull;
	import javax.annotation.Nonnull;
	import javax.annotation.ParametersAreNonnullByDefault;
	import javax.management.NotCompliantMBeanException;

	import org.apache.sling.distribution.journal.impl.event.DistributionEvent;
	import org.apache.sling.distribution.journal.impl.queue.PubQueueProvider;
	import org.apache.sling.distribution.journal.impl.shared.AgentState;
	import org.apache.sling.distribution.journal.impl.shared.DefaultDistributionLog;
	import org.apache.sling.distribution.journal.impl.shared.DistributionMetricsService;
	import org.apache.sling.distribution.journal.impl.shared.SimpleDistributionResponse;
	import org.apache.sling.distribution.journal.impl.shared.Topics;
	import org.apache.commons.lang3.StringUtils;
	import org.apache.sling.api.resource.ResourceResolver;
	import org.apache.sling.distribution.DistributionRequest;
	import org.apache.sling.distribution.DistributionRequestState;
	import org.apache.sling.distribution.DistributionRequestType;
	import org.apache.sling.distribution.DistributionResponse;
	import org.apache.sling.distribution.agent.DistributionAgentState;
	import org.apache.sling.distribution.agent.spi.DistributionAgent;
	import org.apache.sling.distribution.common.DistributionException;
	import org.apache.sling.distribution.log.spi.DistributionLog;
	import org.apache.sling.distribution.packaging.DistributionPackageBuilder;
	import org.apache.sling.distribution.queue.spi.DistributionQueue;
	import org.osgi.framework.BundleContext;
	import org.osgi.framework.ServiceRegistration;
	import org.osgi.service.component.annotations.Activate;
	import org.osgi.service.component.annotations.Component;
	import org.osgi.service.component.annotations.Deactivate;
	import org.osgi.service.component.annotations.Reference;
	import org.osgi.service.event.Event;
	import org.osgi.service.event.EventAdmin;
	import org.osgi.service.metatype.annotations.Designate;

	import org.apache.sling.distribution.journal.impl.shared.JMXRegistration;
	import org.apache.sling.distribution.journal.messages.Messages.PackageMessage;
	import org.apache.sling.distribution.journal.MessageSender;
	import org.apache.sling.distribution.journal.MessagingProvider;
	import org.apache.sling.distribution.journal.JournalAvailable;

	/**
	* A Publisher SCD agent which produces messages to be consumed by a {@code DistributionSubscriber} agent.
	*/
	@Component(
	service = {},
	immediate = true,
	configurationPid = "org.apache.sling.distribution.journal.impl.publisher.DistributionPublisherFactory"
	)
	@Designate(ocd = PublisherConfiguration.class, factory = true)
	@ParametersAreNonnullByDefault
	public class DistributionPublisher implements DistributionAgent {

	private final Map<DistributionRequestType, Consumer<PackageMessage>> REQ_TYPES = new HashMap<>();

	private final DefaultDistributionLog log;

	@Reference
	private MessagingProvider messagingProvider;

	@Reference(name = "packageBuilder")
	private DistributionPackageBuilder packageBuilder;

	@Reference
	private PackageQueuedNotifier queuedNotifier;

	@Reference
	private PubQueueProvider pubQueueProvider;

	@Reference
	private DiscoveryService discoveryService;

	@Reference
	private PackageMessageFactory factory;

	@Reference
	private EventAdmin eventAdmin;

	@Reference
	private Topics topics;

	@Reference
	JournalAvailable journalAvailable;

	@Reference
	private DistributionMetricsService distributionMetricsService;

	private String pubAgentName;

	private String pkgType;

	private long queuedTimeout;

	private ServiceRegistration<DistributionAgent> componentReg;

	private MessageSender<PackageMessage> sender;

	private JMXRegistration reg;

	public DistributionPublisher() {
	log = new DefaultDistributionLog(pubAgentName, this.getClass(), DefaultDistributionLog.LogLevel.INFO);
	REQ_TYPES.put(ADD, this::sendAndWait);
	REQ_TYPES.put(DELETE, this::sendAndWait);
	REQ_TYPES.put(TEST, this::send);
	}

	@Activate
	public void activate(PublisherConfiguration config, BundleContext context) {
	requireNonNull(factory);
	pubAgentName = requireNonNull(config.name());

	queuedTimeout = config.queuedTimeout();

	pkgType = packageBuilder.getType();

	this.sender = messagingProvider.createSender();

	Dictionary<String, Object> props = createServiceProps(config);
	componentReg = requireNonNull(context.registerService(DistributionAgent.class, this, props));

	DistPublisherJMX bean;
	try {
	bean = new DistPublisherJMX(pubAgentName, discoveryService, this);
	} catch (NotCompliantMBeanException e) {
	throw new RuntimeException(e);
	}
	reg = new JMXRegistration(bean, "agent", pubAgentName);

	String msg = String.format("Started Publisher agent %s with packageBuilder %s, queuedTimeout %s",
	pubAgentName, pkgType, queuedTimeout);
	log.info(msg);
	}

	@Deactivate
	public void deactivate() {
	reg.close();
	componentReg.unregister();
	String msg = String.format("Stopped Publisher agent %s with packageBuilder %s, queuedTimeout %s",
	pubAgentName, pkgType, queuedTimeout);
	log.info(msg);
	}

	private Dictionary<String, Object> createServiceProps(PublisherConfiguration config) {
	Dictionary<String, Object> props = new Hashtable<>();
	props.put("name", config.name());
	props.put("title", config.name());
	props.put("details", config.name());
	props.put("packageBuilder.target", config.packageBuilder_target());
	props.put("webconsole.configurationFactory.nameHint", config.webconsole_configurationFactory_nameHint());
	return props;
	}

	/**
	* Get queue names for alive subscribed subscriber agents.
	*/
	@Nonnull
	@Override
	public Iterable<String> getQueueNames() {

	// Queues names are generated only for the subscriber agents which are
	// alive and are subscribed to the publisher agent name (pubAgentName).
	// The queue names match the subscriber agent identifier (subAgentId).
	//
	// If errors queues are enabled, an error queue name is generated which
	// follows the pattern "%s-error". The pattern is deliberately different
	// from the SCD on Jobs one ("error-%s") as we don't want to support
	// the UI ability to retry items from the error queue.
	Set<String> queueNames = new HashSet<>();
	TopologyView view = discoveryService.getTopologyView();
	for (String subAgentId : view.getSubscribedAgentIds(pubAgentName)) {
	queueNames.add(subAgentId);
	State subState = view.getState(subAgentId, pubAgentName);
	if (subState != null) {
	boolean errorQueueEnabled = (subState.getMaxRetries() >= 0);
	if (errorQueueEnabled) {
	queueNames.add(String.format("%s-error", subAgentId));
	}
	}
	}
	return Collections.unmodifiableCollection(queueNames);
	}

	@Override
	public DistributionQueue getQueue(String queueName) {

	// We expect queueName to come from #getQueueNames
	// and this, queueName matches <subAgentId>-error or <subAgentId>
	return queueName.endsWith("-error") ? getErrorQueue(queueName) : getPubQueue(queueName);
	}

	@Nonnull
	private DistributionQueue getErrorQueue(String queueName) {
	AgentId subAgentId = new AgentId(StringUtils.substringBeforeLast(queueName, "-error"));
	return pubQueueProvider.getErrorQueue(pubAgentName, subAgentId.getSlingId(), subAgentId.getAgentName(), queueName);
	}

	@CheckForNull
	private DistributionQueue getPubQueue(String queueName) {
	TopologyView view = discoveryService.getTopologyView();
	AgentId subAgentId = new AgentId(queueName);
	State state = view.getState(subAgentId.getAgentId(), pubAgentName);
	if (state != null) {
	return pubQueueProvider.getQueue(pubAgentName, subAgentId.getSlingId(), subAgentId.getAgentName(), queueName, state.getOffset() + 1, state.getRetries(), state.isEditable());
	}
	return null;
	}



	@Nonnull
	@Override
	public DistributionLog getLog() {
	return log;
	}

	@Nonnull
	@Override
	public DistributionAgentState getState() {
	return AgentState.getState(this);
	}

	@Nonnull
	@Override
	public DistributionResponse execute(ResourceResolver resourceResolver,
	DistributionRequest request)
	throws DistributionException {
	Consumer<PackageMessage> handler = REQ_TYPES.get(request.getRequestType());
	if (handler != null) {
	return execute(resourceResolver, request, handler);
	} else {
	return executeUnsupported(request);
	}
	}

	private DistributionResponse execute(ResourceResolver resourceResolver,
	DistributionRequest request,
	Consumer<PackageMessage> sender)
	throws DistributionException {
	try {
	PackageMessage pkg = timed(distributionMetricsService.getBuildPackageDuration(),
	() -> factory.create(packageBuilder, resourceResolver, pubAgentName, request)
	);
	timed(distributionMetricsService.getEnqueuePackageDuration(),
	() -> sender.accept(pkg)
	);
	distributionMetricsService.getExportedPackageSize().update(pkg.getPkgLength());
	distributionMetricsService.getAcceptedRequests().mark();
	String msg = String.format("Distribution request accepted with type %s paths %s ", request.getRequestType(), Arrays.toString(request.getPaths()));
	log.info(msg);
	return new SimpleDistributionResponse(ACCEPTED, msg);
	} catch (Throwable e) {
	distributionMetricsService.getDroppedRequests().mark();
	String msg = String.format("Failed to queue distribution request %s", e.getMessage());
	log.error(msg, e);
	if (e instanceof Error) {
	throw (Error) e;
	} else {
	throw new DistributionException(msg, e);
	}
	}
	}

	private void sendAndWait(PackageMessage pkg) {
	try {
	CompletableFuture<Void> received = queuedNotifier.registerWait(pkg.getPkgId());
	Event createdEvent = DistributionEvent.eventPackageCreated(pkg, pubAgentName);
	eventAdmin.postEvent(createdEvent);
	send(pkg);
	received.get(queuedTimeout, TimeUnit.MILLISECONDS);
	} catch (Exception e) {
	queuedNotifier.unRegisterWait(pkg.getPkgId());
	throw new RuntimeException(e);
	}
	}

	private void send(PackageMessage pipePackage) {
	final String topicName = topics.getPackageTopic();
	sender.send(topicName, pipePackage);
	}

	@Nonnull
	private DistributionResponse executeUnsupported(DistributionRequest request) {
	String msg = String.format("Request type %s is not supported by this agent, expected one of %s",
	request.getRequestType(), REQ_TYPES.keySet());
	log.info(msg);
	return new SimpleDistributionResponse(DistributionRequestState.DROPPED, msg);
	}
	}