PublishEngageWorkflowOperationHandler.java
/*
* Licensed to The Apereo Foundation under one or more contributor license
* agreements. See the NOTICE file distributed with this work for additional
* information regarding copyright ownership.
*
*
* The Apereo Foundation licenses this file to you under the Educational
* Community 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://opensource.org/licenses/ecl2.txt
*
* 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.opencastproject.workflow.handler.distribution;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.opencastproject.systems.OpencastConstants.SERVER_URL_PROPERTY;
import static org.opencastproject.util.data.Option.option;
import static org.opencastproject.util.data.functions.Strings.toBool;
import static org.opencastproject.util.data.functions.Strings.trimToNone;
import static org.opencastproject.workflow.handler.distribution.EngagePublicationChannel.CHANNEL_ID;
import org.opencastproject.distribution.api.DistributionException;
import org.opencastproject.distribution.api.DownloadDistributionService;
import org.opencastproject.distribution.api.StreamingDistributionService;
import org.opencastproject.job.api.Job;
import org.opencastproject.job.api.JobContext;
import org.opencastproject.mediapackage.Attachment;
import org.opencastproject.mediapackage.Catalog;
import org.opencastproject.mediapackage.MediaPackage;
import org.opencastproject.mediapackage.MediaPackageElement;
import org.opencastproject.mediapackage.MediaPackageElementFlavor;
import org.opencastproject.mediapackage.MediaPackageElementParser;
import org.opencastproject.mediapackage.MediaPackageElements;
import org.opencastproject.mediapackage.MediaPackageException;
import org.opencastproject.mediapackage.MediaPackageReference;
import org.opencastproject.mediapackage.MediaPackageReferenceImpl;
import org.opencastproject.mediapackage.Publication;
import org.opencastproject.mediapackage.PublicationImpl;
import org.opencastproject.mediapackage.Track;
import org.opencastproject.mediapackage.selector.SimpleElementSelector;
import org.opencastproject.mediapackage.track.TrackImpl;
import org.opencastproject.metadata.dublincore.DublinCore;
import org.opencastproject.metadata.dublincore.DublinCoreValue;
import org.opencastproject.metadata.dublincore.DublinCoreXmlFormat;
import org.opencastproject.search.api.SearchException;
import org.opencastproject.search.api.SearchService;
import org.opencastproject.security.api.Organization;
import org.opencastproject.security.api.OrganizationDirectoryService;
import org.opencastproject.security.api.UnauthorizedException;
import org.opencastproject.serviceregistry.api.ServiceRegistry;
import org.opencastproject.serviceregistry.api.ServiceRegistryException;
import org.opencastproject.util.MimeTypes;
import org.opencastproject.util.NotFoundException;
import org.opencastproject.util.UrlSupport;
import org.opencastproject.workflow.api.AbstractWorkflowOperationHandler;
import org.opencastproject.workflow.api.WorkflowInstance;
import org.opencastproject.workflow.api.WorkflowOperationException;
import org.opencastproject.workflow.api.WorkflowOperationHandler;
import org.opencastproject.workflow.api.WorkflowOperationInstance;
import org.opencastproject.workflow.api.WorkflowOperationResult;
import org.opencastproject.workflow.api.WorkflowOperationResult.Action;
import org.opencastproject.workspace.api.Workspace;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.client.utils.URIUtils;
import org.osgi.framework.BundleContext;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Reference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;
/**
* The workflow definition for handling "engage publication" operations
*/
@Component(
immediate = true,
service = WorkflowOperationHandler.class,
property = {
"service.description=Engage Publication Workflow Handler",
"workflow.operation=publish-engage"
}
)
public class PublishEngageWorkflowOperationHandler extends AbstractWorkflowOperationHandler {
/** The logging facility */
private static final Logger logger = LoggerFactory.getLogger(PublishEngageWorkflowOperationHandler.class);
/** Configuration properties id */
static final String ENGAGE_URL_PROPERTY = "org.opencastproject.engage.ui.url";
static final String STREAMING_PUBLISH_PROPERTY = "org.opencastproject.publish.streaming.formats";
/** Workflow configuration option keys */
static final String DOWNLOAD_SOURCE_FLAVORS = "download-source-flavors";
static final String DOWNLOAD_TARGET_SUBFLAVOR = "download-target-subflavor";
static final String DOWNLOAD_SOURCE_TAGS = "download-source-tags";
static final String DOWNLOAD_TARGET_TAGS = "download-target-tags";
static final String STREAMING_SOURCE_TAGS = "streaming-source-tags";
static final String STREAMING_TARGET_TAGS = "streaming-target-tags";
static final String STREAMING_SOURCE_FLAVORS = "streaming-source-flavors";
static final String STREAMING_TARGET_SUBFLAVOR = "streaming-target-subflavor";
static final String CHECK_AVAILABILITY = "check-availability";
static final String STRATEGY = "strategy";
static final String MERGE_FORCE_FLAVORS = "merge-force-flavors";
static final String ADD_FORCE_FLAVORS = "add-force-flavors";
private static final String MERGE_FORCE_FLAVORS_DEFAULT = "dublincore/*,security/*";
private static final String ADD_FORCE_FLAVORS_DEFAULT = "";
/** Path the REST endpoint which will re-direct users to the currently configured video player **/
static final String PLAYER_PATH = "/play/";
/** Name constant for the 'merge' strategy **/
static final String PUBLISH_STRATEGY_MERGE = "merge";
/** Name constant for the 'default' 'strategy **/
static final String PUBLISH_STRATEGY_DEFAULT = "default";
/** The streaming distribution service */
private StreamingDistributionService streamingDistributionService = null;
/** The download distribution service */
private DownloadDistributionService downloadDistributionService = null;
/** The search service */
private SearchService searchService = null;
private Workspace workspace;
/** The server url */
private URL serverUrl;
private OrganizationDirectoryService organizationDirectoryService = null;
/** Which streaming formats should be published automatically */
private List<String> publishedStreamingFormats = null;
/**
* Callback for the OSGi declarative services configuration.
*
* @param streamingDistributionService
* the streaming distribution service
*/
@Reference(target = "(distribution.channel=streaming)")
protected void setStreamingDistributionService(StreamingDistributionService streamingDistributionService) {
this.streamingDistributionService = streamingDistributionService;
}
/**
* Callback for the OSGi declarative services configuration.
*
* @param downloadDistributionService
* the download distribution service
*/
@Reference(target = "(distribution.channel=download)")
protected void setDownloadDistributionService(DownloadDistributionService downloadDistributionService) {
this.downloadDistributionService = downloadDistributionService;
}
/**
* Callback for declarative services configuration that will introduce us to the search service. Implementation
* assumes that the reference is configured as being static.
*
* @param searchService
* an instance of the search service
*/
@Reference
protected void setSearchService(SearchService searchService) {
this.searchService = searchService;
}
@Reference
public void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectoryService) {
this.organizationDirectoryService = organizationDirectoryService;
}
@Reference
@Override
public void setServiceRegistry(ServiceRegistry serviceRegistry) {
super.setServiceRegistry(serviceRegistry);
}
@Reference
public void setWorkspace(Workspace workspace) {
this.workspace = workspace;
}
/** Supported streaming formats */
private static final Set<TrackImpl.StreamingProtocol> STREAMING_FORMATS = new HashSet<>(Arrays.asList(
TrackImpl.StreamingProtocol.RTMP,
TrackImpl.StreamingProtocol.RTMPE,
TrackImpl.StreamingProtocol.HLS,
TrackImpl.StreamingProtocol.DASH,
TrackImpl.StreamingProtocol.HDS,
TrackImpl.StreamingProtocol.SMOOTH));
@Override
@Activate
protected void activate(ComponentContext cc) {
super.activate(cc);
BundleContext bundleContext = cc.getBundleContext();
// Get configuration
serverUrl = UrlSupport.url(bundleContext.getProperty(SERVER_URL_PROPERTY));
publishedStreamingFormats = Arrays.asList(Optional.ofNullable(StringUtils.split(
bundleContext.getProperty(STREAMING_PUBLISH_PROPERTY), ",")).orElse(new String[0]));
}
@Override
public WorkflowOperationResult start(final WorkflowInstance workflowInstance, JobContext context)
throws WorkflowOperationException {
logger.debug("Running engage publication workflow operation");
MediaPackage mediaPackage = workflowInstance.getMediaPackage();
WorkflowOperationInstance op = workflowInstance.getCurrentOperation();
// Check which tags have been configured
String downloadSourceTags = StringUtils.trimToEmpty(op.getConfiguration(DOWNLOAD_SOURCE_TAGS));
String downloadTargetTags = StringUtils.trimToEmpty(op.getConfiguration(DOWNLOAD_TARGET_TAGS));
String downloadSourceFlavors = StringUtils.trimToEmpty(op.getConfiguration(DOWNLOAD_SOURCE_FLAVORS));
String downloadTargetSubflavor = StringUtils.trimToNull(op.getConfiguration(DOWNLOAD_TARGET_SUBFLAVOR));
String streamingSourceTags = StringUtils.trimToEmpty(op.getConfiguration(STREAMING_SOURCE_TAGS));
String streamingTargetTags = StringUtils.trimToEmpty(op.getConfiguration(STREAMING_TARGET_TAGS));
String streamingSourceFlavors = StringUtils.trimToEmpty(op.getConfiguration(STREAMING_SOURCE_FLAVORS));
String streamingTargetSubflavor = StringUtils.trimToNull(op.getConfiguration(STREAMING_TARGET_SUBFLAVOR));
String republishStrategy = StringUtils.trimToEmpty(
StringUtils.defaultString(op.getConfiguration(STRATEGY), PUBLISH_STRATEGY_DEFAULT));
String mergeForceFlavorsStr = StringUtils.trimToEmpty(
StringUtils.defaultString(op.getConfiguration(MERGE_FORCE_FLAVORS), MERGE_FORCE_FLAVORS_DEFAULT));
String addForceFlavorsStr = StringUtils.trimToEmpty(
StringUtils.defaultString(op.getConfiguration(ADD_FORCE_FLAVORS), ADD_FORCE_FLAVORS_DEFAULT));
boolean checkAvailability = option(op.getConfiguration(CHECK_AVAILABILITY)).bind(trimToNone).map(toBool)
.getOrElse(true);
// First check if mp exists in the search index and strategy is merge
// to avoid leaving distributed elements around.
MediaPackage distributedMp = null;
try {
distributedMp = searchService.get(mediaPackage.getIdentifier().toString());
} catch (NotFoundException e) {
logger.debug("No published mediapackage found for {}", mediaPackage.getIdentifier().toString());
} catch (UnauthorizedException e) {
throw new WorkflowOperationException("Unauthorized for " + mediaPackage.getIdentifier().toString(), e);
}
if (PUBLISH_STRATEGY_MERGE.equals(republishStrategy) && distributedMp == null) {
logger.info("Skipping republish for {} since it is not currently published",
mediaPackage.getIdentifier().toString());
return createResult(mediaPackage, Action.SKIP);
}
String[] sourceDownloadTags = StringUtils.split(downloadSourceTags, ",");
String[] targetDownloadTags = StringUtils.split(downloadTargetTags, ",");
String[] sourceDownloadFlavors = StringUtils.split(downloadSourceFlavors, ",");
String[] sourceStreamingTags = StringUtils.split(streamingSourceTags, ",");
String[] targetStreamingTags = StringUtils.split(streamingTargetTags, ",");
String[] sourceStreamingFlavors = StringUtils.split(streamingSourceFlavors, ",");
if (sourceDownloadTags.length == 0 && sourceDownloadFlavors.length == 0 && sourceStreamingTags.length == 0
&& sourceStreamingFlavors.length == 0) {
logger.warn("No tags or flavors have been specified, so nothing will be published to the "
+ "engage publication channel");
return createResult(mediaPackage, Action.CONTINUE);
}
// Parse forced flavors
List<MediaPackageElementFlavor> mergeForceFlavors = Arrays.stream(StringUtils.split(mergeForceFlavorsStr, ", "))
.map(MediaPackageElementFlavor::parseFlavor).collect(Collectors.toList());
List<MediaPackageElementFlavor> addForceFlavors = Arrays.stream(StringUtils.split(addForceFlavorsStr, ", "))
.map(MediaPackageElementFlavor::parseFlavor).collect(Collectors.toList());
// Parse the download target flavor
MediaPackageElementFlavor downloadSubflavor = null;
if (downloadTargetSubflavor != null) {
try {
downloadSubflavor = MediaPackageElementFlavor.parseFlavor(downloadTargetSubflavor);
} catch (IllegalArgumentException e) {
throw new WorkflowOperationException(e);
}
}
// Parse the streaming target flavor
MediaPackageElementFlavor streamingSubflavor = null;
if (streamingTargetSubflavor != null) {
try {
streamingSubflavor = MediaPackageElementFlavor.parseFlavor(streamingTargetSubflavor);
} catch (IllegalArgumentException e) {
throw new WorkflowOperationException(e);
}
}
// Configure the download element selector
SimpleElementSelector downloadElementSelector = new SimpleElementSelector();
for (String flavor : sourceDownloadFlavors) {
downloadElementSelector.addFlavor(MediaPackageElementFlavor.parseFlavor(flavor));
}
for (String tag : sourceDownloadTags) {
downloadElementSelector.addTag(tag);
}
// Configure the streaming element selector
SimpleElementSelector streamingElementSelector = new SimpleElementSelector();
for (String flavor : sourceStreamingFlavors) {
streamingElementSelector.addFlavor(MediaPackageElementFlavor.parseFlavor(flavor));
}
for (String tag : sourceStreamingTags) {
streamingElementSelector.addTag(tag);
}
// Select the appropriate elements for download and streaming
Collection<MediaPackageElement> downloadElements = downloadElementSelector.select(mediaPackage, false);
Collection<MediaPackageElement> streamingElements = streamingElementSelector.select(mediaPackage, false);
try {
Set<String> downloadElementIds = new HashSet<String>();
Set<String> streamingElementIds = new HashSet<String>();
// Look for elements matching the tag
for (MediaPackageElement elem : downloadElements) {
downloadElementIds.add(elem.getIdentifier());
}
for (MediaPackageElement elem : streamingElements) {
streamingElementIds.add(elem.getIdentifier());
}
removePublicationElement(mediaPackage);
if (republishStrategy.equals(PUBLISH_STRATEGY_DEFAULT)) {
retractFromEngage(distributedMp);
}
List<Job> jobs = new ArrayList<Job>();
//distribute Elements
try {
if (downloadElementIds.size() > 0) {
Job job = downloadDistributionService.distribute(
CHANNEL_ID, mediaPackage, downloadElementIds, checkAvailability);
if (job != null) {
jobs.add(job);
}
}
if (streamingDistributionService != null && streamingDistributionService.publishToStreaming()) {
for (String elementId : streamingElementIds) {
Job job = streamingDistributionService.distribute(CHANNEL_ID, mediaPackage, elementId);
if (job != null) {
jobs.add(job);
}
}
}
} catch (DistributionException e) {
throw new WorkflowOperationException(e);
}
if (jobs.size() < 1) {
logger.info("No mediapackage element was found for distribution to engage");
return createResult(mediaPackage, Action.CONTINUE);
}
// Wait until all distribution jobs have returned
if (!waitForStatus(jobs.toArray(new Job[jobs.size()])).isSuccess()) {
throw new WorkflowOperationException("One of the distribution jobs did not complete successfully");
}
logger.debug("Distribute of mediapackage {} completed", mediaPackage);
String engageUrlString = null;
try {
MediaPackage mediaPackageForSearch = getMediaPackageForSearchIndex(mediaPackage, jobs, downloadSubflavor,
targetDownloadTags, downloadElementIds, streamingSubflavor, streamingElementIds, targetStreamingTags);
// MH-10216, check if only merging into existing mediapackage
removePublicationElement(mediaPackage);
if (republishStrategy.equals(PUBLISH_STRATEGY_MERGE)) {
mediaPackageForSearch = mergePackages(mediaPackageForSearch, distributedMp, mergeForceFlavors,
addForceFlavors);
}
if (StringUtils.isBlank(mediaPackageForSearch.getTitle())) {
var dcUri = Arrays.stream(mediaPackageForSearch.getCatalogs(MediaPackageElements.EPISODE))
.findFirst()
.map(MediaPackageElement::getURI);
if (dcUri.isPresent()) {
try (var in = workspace.read(dcUri.get())) {
DublinCoreXmlFormat.read(in)
.get(DublinCore.PROPERTY_TITLE)
.stream()
.findFirst()
.map(DublinCoreValue::getValue)
.ifPresent(mediaPackageForSearch::setTitle);
}
}
}
// Check that the media package meets the criteria for publication
if (isBlank(mediaPackageForSearch.getTitle())) {
throw new WorkflowOperationException("Media package does not meet publication criteria: Missing title");
}
if (!mediaPackageForSearch.hasTracks()) {
throw new WorkflowOperationException("Media package does not meet publication criteria: No tracks selected");
}
// Prepare published elements to be added
MediaPackageElement[] mediaPackageElements = mediaPackageForSearch.getElements();
logger.info("Publishing media package {} to search index", mediaPackageForSearch);
URL engageBaseUrl;
Organization organization = organizationDirectoryService.getOrganization(workflowInstance.getOrganizationId());
engageUrlString = StringUtils.trimToNull(organization.getProperties().get(ENGAGE_URL_PROPERTY));
if (engageUrlString != null) {
engageBaseUrl = new URL(engageUrlString);
} else {
engageBaseUrl = serverUrl;
logger.info(
"Using 'server.url' as a fallback for the non-existing organization level key '{}' "
+ "for the publication url",
ENGAGE_URL_PROPERTY);
}
// create the publication URI (used by Admin UI for event details link)
URI engageUri = this.createEngageUri(engageBaseUrl.toURI(), mediaPackage);
// Create new distribution element
Publication publicationElement = PublicationImpl.publication(UUID.randomUUID().toString(), CHANNEL_ID,
engageUri, MimeTypes.parseMimeType("text/html"));
// Add published elements
for (MediaPackageElement element : mediaPackageElements) {
element.setIdentifier(null);
PublicationImpl.addElementToPublication(publicationElement, element);
}
mediaPackage.add(publicationElement);
// create publication URI for streaming
if (streamingDistributionService != null
&& streamingDistributionService.publishToStreaming()
&& !publishedStreamingFormats.isEmpty()) {
for (Track track : mediaPackageForSearch.getTracks()) {
String mimeType = track.getMimeType().toString();
if (isStreamingFormat(track) && (publishedStreamingFormats.contains(mimeType)
|| publishedStreamingFormats.contains("*"))) {
publicationElement.addTrack(track);
}
}
for (Attachment attachment : mediaPackageForSearch.getAttachments()) {
publicationElement.addAttachment(attachment);
}
for (Catalog catalog : mediaPackageForSearch.getCatalogs()) {
publicationElement.addCatalog(catalog);
}
}
// Adding media package to the search index
Job publishJob = null;
try {
publishJob = searchService.add(mediaPackageForSearch);
if (!waitForStatus(publishJob).isSuccess()) {
throw new WorkflowOperationException("Mediapackage " + mediaPackageForSearch.getIdentifier()
+ " could not be published");
}
} catch (SearchException e) {
throw new WorkflowOperationException("Error publishing media package", e);
} catch (MediaPackageException e) {
throw new WorkflowOperationException("Error parsing media package", e);
}
logger.debug("Publishing of mediapackage {} completed", mediaPackage);
return createResult(mediaPackage, Action.CONTINUE);
} catch (MalformedURLException e) {
logger.error("{} is malformed: {}", ENGAGE_URL_PROPERTY, engageUrlString);
throw new WorkflowOperationException(e);
} catch (Throwable t) {
if (t instanceof WorkflowOperationException) {
throw (WorkflowOperationException) t;
} else {
throw new WorkflowOperationException(t);
}
}
} catch (Exception e) {
if (e instanceof WorkflowOperationException) {
throw (WorkflowOperationException) e;
} else {
throw new WorkflowOperationException(e);
}
}
}
/**
* Local utility to assemble player path for this class
*
* @param engageUri
* @param mp
* @return the assembled player URI for this mediapackage
*/
URI createEngageUri(URI engageUri, MediaPackage mp) {
return URIUtils.resolve(engageUri, PLAYER_PATH + mp.getIdentifier().toString());
}
/**
* Returns a mediapackage that only contains elements that are marked for distribution.
*
* @param current
* the current mediapackage
* @param jobs
* the distribution jobs
* @param downloadSubflavor
* flavor to be applied to elements distributed to download
* @param downloadTargetTags
* tags to be applied to elements distributed to downloads
* @param downloadElementIds
* identifiers for elements that have been distributed to downloads
* @param streamingSubflavor
* flavor to be applied to elements distributed to streaming
* @param streamingElementIds
* identifiers for elements that have been distributed to streaming
* @param streamingTargetTags
* tags to be applied to elements distributed to streaming
* @return the new mediapackage
*/
protected MediaPackage getMediaPackageForSearchIndex(MediaPackage current, List<Job> jobs,
MediaPackageElementFlavor downloadSubflavor, String[] downloadTargetTags, Set<String> downloadElementIds,
MediaPackageElementFlavor streamingSubflavor, Set<String> streamingElementIds, String[] streamingTargetTags)
throws MediaPackageException, NotFoundException, ServiceRegistryException, WorkflowOperationException {
MediaPackage mp = (MediaPackage) current.clone();
// All the jobs have passed, let's update the mediapackage with references to the distributed elements
List<String> elementsToPublish = new ArrayList<String>();
Map<String, String> distributedElementIds = new HashMap<String, String>();
for (Job entry : jobs) {
Job job = serviceRegistry.getJob(entry.getId());
// If there is no payload, then the item has not been distributed.
if (job.getPayload() == null) {
continue;
}
List<? extends MediaPackageElement> distributedElements = null;
try {
distributedElements = MediaPackageElementParser.getArrayFromXml(job.getPayload());
} catch (MediaPackageException e) {
throw new WorkflowOperationException(e);
}
// If the job finished successfully, but returned no new element, the channel simply doesn't support this
// kind of element. So we just keep on looping.
if (distributedElements == null || distributedElements.size() < 1) {
continue;
}
for (MediaPackageElement distributedElement : distributedElements) {
String sourceElementId = distributedElement.getIdentifier();
if (sourceElementId != null) {
MediaPackageElement sourceElement = mp.getElementById(sourceElementId);
// Make sure the mediapackage is prompted to create a new identifier for this element
distributedElement.setIdentifier(null);
if (sourceElement != null) {
// Adjust the flavor and tags for downloadable elements
if (downloadElementIds.contains(sourceElementId)) {
if (downloadSubflavor != null) {
MediaPackageElementFlavor flavor = sourceElement.getFlavor();
if (flavor != null) {
MediaPackageElementFlavor newFlavor = new MediaPackageElementFlavor(flavor.getType(),
downloadSubflavor.getSubtype());
distributedElement.setFlavor(newFlavor);
}
}
}
// Adjust the flavor and tags for streaming elements
else if (streamingElementIds.contains(sourceElementId)) {
if (streamingSubflavor != null && streamingElementIds.contains(sourceElementId)) {
MediaPackageElementFlavor flavor = sourceElement.getFlavor();
if (flavor != null) {
MediaPackageElementFlavor newFlavor = new MediaPackageElementFlavor(flavor.getType(),
streamingSubflavor.getSubtype());
distributedElement.setFlavor(newFlavor);
}
}
}
// Copy references from the source elements to the distributed elements
MediaPackageReference ref = sourceElement.getReference();
if (ref != null && mp.getElementByReference(ref) != null) {
MediaPackageReference newReference = (MediaPackageReference) ref.clone();
distributedElement.setReference(newReference);
}
}
}
if (isStreamingFormat(distributedElement)) {
applyTags(distributedElement, streamingTargetTags);
} else {
applyTags(distributedElement, downloadTargetTags);
}
// Add the new element to the mediapackage
mp.add(distributedElement);
elementsToPublish.add(distributedElement.getIdentifier());
distributedElementIds.put(sourceElementId, distributedElement.getIdentifier());
}
}
// Mark everything that is set for removal
List<MediaPackageElement> removals = new ArrayList<MediaPackageElement>();
for (MediaPackageElement element : mp.getElements()) {
if (!elementsToPublish.contains(element.getIdentifier())) {
removals.add(element);
}
}
// Translate references to the distributed artifacts
for (MediaPackageElement element : mp.getElements()) {
if (removals.contains(element)) {
continue;
}
// Is the element referencing anything?
MediaPackageReference reference = element.getReference();
if (reference == null) {
continue;
}
// See if the element has been distributed
String distributedElementId = distributedElementIds.get(reference.getIdentifier());
if (distributedElementId == null) {
continue;
}
MediaPackageReference translatedReference
= new MediaPackageReferenceImpl(mp.getElementById(distributedElementId));
if (reference.getProperties() != null) {
translatedReference.getProperties().putAll(reference.getProperties());
}
// Set the new reference
element.setReference(translatedReference);
}
// Remove everything we don't want to add to publish
for (MediaPackageElement element : removals) {
mp.remove(element);
}
return mp;
}
/**
* Checks if the MediaPackage track transport protocol is a streaming format protocol
* @param element The MediaPackageElement to analyze
* @return true if it is a TrackImpl and has a streaming protocol as transport
*/
private boolean isStreamingFormat(MediaPackageElement element) {
return element instanceof TrackImpl
&& STREAMING_FORMATS.contains(((TrackImpl) element).getTransport());
}
/**
* Adds Tags to a MediaPackageElement
* @param element the element that needs the tags
* @param tags the list of tags to apply
*/
private void applyTags(MediaPackageElement element, String[] tags) {
for (String tag : tags) {
element.addTag(tag);
}
}
/**
* MH-10216, Copied from the original RepublishWorkflowOperationHandler
*
* Merges the updated mediapackage with the one that is currently published in a way where the updated elements
* replace existing ones in the published mediapackage based on their flavor.
* <p>
* If <code>publishedMp</code> is <code>null</code>, this method returns the updated mediapackage without any
* modifications.
*
* @param updatedMp
* the updated media package
* @param publishedMp
* the mediapackage that is currently published
* @return the merged mediapackage
*/
protected MediaPackage mergePackages(MediaPackage updatedMp, MediaPackage publishedMp,
List<MediaPackageElementFlavor> mergeForceFlavors, List<MediaPackageElementFlavor> addForceFlavors) {
if (publishedMp == null) {
return updatedMp;
}
MediaPackage mergedMediaPackage = (MediaPackage) updatedMp.clone();
for (MediaPackageElement element : publishedMp.elements()) {
String type = element.getElementType().toString().toLowerCase();
boolean elementHasFlavorThatAlreadyExists = updatedMp.getElementsByFlavor(element.getFlavor()).length > 0;
boolean elementHasForceMergeFlavor = mergeForceFlavors.stream().anyMatch((f) -> element.getFlavor().matches(f));
boolean elementHasForceAddFlavor = addForceFlavors.stream().anyMatch((f) -> element.getFlavor().matches(f));
if (elementHasForceAddFlavor) {
logger.info("Adding {} '{}' into the updated mediapackage", type, element.getIdentifier());
mergedMediaPackage.add((MediaPackageElement) element.clone());
continue;
}
if (!elementHasFlavorThatAlreadyExists) {
if (elementHasForceMergeFlavor) {
logger.info("Forcing removal of {} {} due to the absence of a new element with flavor {}",
type, element.getIdentifier(), element.getFlavor().toString());
continue;
}
logger.info("Merging {} '{}' into the updated mediapackage", type, element.getIdentifier());
mergedMediaPackage.add((MediaPackageElement) element.clone());
} else {
logger.info("Overwriting existing {} '{}' with '{}' in the updated mediapackage",
type, element.getIdentifier(), updatedMp.getElementsByFlavor(element.getFlavor())[0].getIdentifier());
}
}
return mergedMediaPackage;
}
private void removePublicationElement(MediaPackage mediaPackage) {
for (Publication publicationElement : mediaPackage.getPublications()) {
if (CHANNEL_ID.equals(publicationElement.getChannel())) {
mediaPackage.remove(publicationElement);
}
}
}
/**
* Removes Mediapackage from Searchindex
*
* @param distributedMediaPackage
* The media package gotten from the search index
* @throws WorkflowOperationException
*/
private void retractFromEngage(MediaPackage distributedMediaPackage) throws WorkflowOperationException {
List<Job> jobs = new ArrayList<>();
Set<String> elementIds = new HashSet<>();
try {
if (distributedMediaPackage != null) {
for (MediaPackageElement element : distributedMediaPackage.getElements()) {
elementIds.add(element.getIdentifier());
}
//bulk retraction
if (elementIds.size() > 0) {
Job retractDownloadDistributionJob
= downloadDistributionService.retract(CHANNEL_ID, distributedMediaPackage, elementIds);
if (retractDownloadDistributionJob != null) {
jobs.add(retractDownloadDistributionJob);
}
}
if (streamingDistributionService != null && streamingDistributionService.publishToStreaming()) {
for (MediaPackageElement element : distributedMediaPackage.getElements()) {
Job retractStreamingJob
= streamingDistributionService.retract(CHANNEL_ID, distributedMediaPackage, element.getIdentifier());
if (retractStreamingJob != null) {
jobs.add(retractStreamingJob);
}
}
}
Job deleteSearchJob = null;
logger.info("Retracting already published Elements for Mediapackage: {}",
distributedMediaPackage.getIdentifier().toString());
deleteSearchJob = searchService.delete(distributedMediaPackage.getIdentifier().toString());
if (deleteSearchJob != null) {
jobs.add(deleteSearchJob);
}
}
// Wait until all retraction jobs have returned
if (!waitForStatus(jobs.toArray(new Job[jobs.size()])).isSuccess()) {
throw new WorkflowOperationException("One of the retraction jobs did not complete successfully");
}
} catch (DistributionException e) {
throw new WorkflowOperationException(e);
} catch (SearchException e) {
throw new WorkflowOperationException("Error retracting media package", e);
} catch (UnauthorizedException | NotFoundException ex) {
logger.error("Retraction failed of Mediapackage: {}", distributedMediaPackage.getIdentifier().toString(), ex);
}
}
}