001/* 002 * (C) Copyright 2018 Nuxeo (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * Funsho David 018 */ 019 020package org.nuxeo.ecm.core.bulk.action; 021 022import static org.nuxeo.ecm.core.bulk.BulkServiceImpl.STATUS_STREAM; 023import static org.nuxeo.lib.stream.computation.AbstractComputation.INPUT_1; 024import static org.nuxeo.lib.stream.computation.AbstractComputation.OUTPUT_1; 025 026import java.io.Serializable; 027import java.util.Arrays; 028import java.util.List; 029import java.util.Map; 030import java.util.Map.Entry; 031 032import org.apache.logging.log4j.LogManager; 033import org.apache.logging.log4j.Logger; 034import org.nuxeo.ecm.core.api.CoreSession; 035import org.nuxeo.ecm.core.api.DocumentModel; 036import org.nuxeo.ecm.core.api.PropertyException; 037import org.nuxeo.ecm.core.api.VersioningOption; 038import org.nuxeo.ecm.core.api.versioning.VersioningService; 039import org.nuxeo.ecm.core.bulk.action.computation.AbstractBulkComputation; 040import org.nuxeo.ecm.core.bulk.message.BulkCommand; 041import org.nuxeo.lib.stream.computation.Topology; 042import org.nuxeo.runtime.stream.StreamProcessorTopology; 043 044/** 045 * @since 10.2 046 */ 047public class SetPropertiesAction implements StreamProcessorTopology { 048 049 public static final String ACTION_NAME = "setProperties"; 050 051 public static final String ACTION_FULL_NAME = "bulk/" + ACTION_NAME; 052 053 // duplicated from NXAuditEventsService.DISABLE_AUDIT_LOGGER 054 public static final String PARAM_DISABLE_AUDIT = "disableAuditLogger"; 055 056 public static final String PARAM_VERSIONING_OPTION = VersioningService.VERSIONING_OPTION; 057 058 @Override 059 public Topology getTopology(Map<String, String> options) { 060 return Topology.builder() 061 .addComputation(SetPropertyComputation::new, 062 Arrays.asList(INPUT_1 + ":" + ACTION_FULL_NAME, // 063 OUTPUT_1 + ":" + STATUS_STREAM)) 064 .build(); 065 } 066 067 public static class SetPropertyComputation extends AbstractBulkComputation { 068 069 private static final Logger log = LogManager.getLogger(SetPropertyComputation.class); 070 071 protected VersioningOption versioningOption; 072 073 protected boolean disableAudit; 074 075 public SetPropertyComputation() { 076 super(ACTION_FULL_NAME); 077 } 078 079 @Override 080 public void startBucket(String bucketKey) { 081 BulkCommand command = getCurrentCommand(); 082 Serializable auditParam = command.getParam(PARAM_DISABLE_AUDIT); 083 disableAudit = auditParam != null && Boolean.parseBoolean(auditParam.toString()); 084 Serializable versioningParam = command.getParam(PARAM_VERSIONING_OPTION); 085 versioningOption = VersioningOption.NONE.toString().equals(versioningParam) ? VersioningOption.NONE : null; 086 } 087 088 @Override 089 protected void compute(CoreSession session, List<String> ids, Map<String, Serializable> properties) { 090 for (DocumentModel doc : loadDocuments(session, ids)) { 091 if (disableAudit) { 092 doc.putContextData(PARAM_DISABLE_AUDIT, Boolean.TRUE); 093 } 094 if (versioningOption != null) { 095 doc.putContextData(VersioningService.VERSIONING_OPTION, versioningOption); 096 } 097 // update properties 098 for (Entry<String, Serializable> es : properties.entrySet()) { 099 if (!PARAM_DISABLE_AUDIT.equals(es.getKey()) && !PARAM_VERSIONING_OPTION.equals(es.getKey())) { 100 try { 101 doc.setPropertyValue(es.getKey(), es.getValue()); 102 } catch (PropertyException e) { 103 // TODO send to error stream 104 log.warn("Cannot write property: {} of document: {}", es.getKey(), doc.getId(), e); 105 } 106 } 107 } 108 try { 109 session.saveDocument(doc); 110 } catch (PropertyException e) { 111 // TODO send to error stream 112 log.warn("Cannot save document: {}", doc.getId(), e); 113 } 114 } 115 } 116 } 117 118}