markap14 commented on code in PR #8120: URL: https://github.com/apache/nifi/pull/8120#discussion_r1414495581
########## nifi-nar-bundles/nifi-slack-bundle/nifi-slack-processors/src/main/java/org/apache/nifi/processors/slack/PublishSlack.java: ########## @@ -0,0 +1,513 @@ +/* + * 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.nifi.processors.slack; + +import com.slack.api.bolt.App; +import com.slack.api.bolt.AppConfig; +import com.slack.api.methods.MethodsClient; +import com.slack.api.methods.request.chat.ChatPostMessageRequest; +import com.slack.api.methods.request.files.FilesUploadV2Request; +import com.slack.api.methods.response.chat.ChatPostMessageResponse; +import com.slack.api.methods.response.files.FilesUploadV2Response; +import com.slack.api.model.File; +import com.slack.api.model.File.ShareDetail; +import com.slack.api.model.File.Shares; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.configuration.DefaultSettings; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.MultiProcessorUseCase; +import org.apache.nifi.annotation.documentation.ProcessorConfiguration; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.documentation.UseCase; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.Validator; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.DataUnit; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processors.slack.consume.ConsumeSlackUtil; +import org.apache.nifi.processors.slack.util.ChannelMapper; +import org.apache.nifi.processors.slack.util.RateLimit; +import org.apache.nifi.stream.io.StreamUtils; +import org.apache.nifi.util.FormatUtils; + +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +@InputRequirement(Requirement.INPUT_REQUIRED) +@CapabilityDescription(""" + Posts a message to the specified Slack channel. The content of the message can be either a user-defined message that makes use of Expression Language or + the contents of the FlowFile can be sent as the message. If sending a user-defined message, the contents of the FlowFile may also be optionally uploaded as + a file attachment. + """) +@SeeAlso({ListenSlack.class, ConsumeSlack.class}) +@Tags({"slack", "conversation", "chat.postMessage", "social media", "team", "text", "unstructured", "write", "upload", "send", "publish"}) +@WritesAttributes({ + @WritesAttribute(attribute = "slack.channel.id", description = "The ID of the Slack Channel from which the messages were retrieved"), + @WritesAttribute(attribute = "slack.ts", description = "The timestamp of the slack messages that was sent; this is used by Slack as a unique identifier") +}) +@DefaultSettings(yieldDuration = "3 sec") +@UseCase( + description = "Send specific text as a message to Slack, optionally including the FlowFile's contents as an attached file.", + configuration = """ + Set "Access Token" to the value of your Slack OAuth Access Token. + Set "Channel" to the ID of the channel or the name of the channel prefixed with the # symbol. For example, "C0123456789" or "#general". + Set "Publish Strategy" to "Use 'Message Text' Property". + Set "Message Text" to the text that you would like to send as the Slack message. + Set "Include FlowFile Content as Attachment" to "true" if the FlowFile's contents should be attached as a file, or "false" to send just the message text without an attachment. + """ +) +@UseCase( + description = "Send the contents of the FlowFile as a message to Slack.", + configuration = """ + Set "Access Token" to the value of your Slack OAuth Access Token. + Set "Channel" to the ID of the channel or the name of the channel prefixed with the # symbol. For example, "C0123456789" or "#general". + Set "Publish Strategy" to "Send FlowFile Content as Message". + """ +) +@MultiProcessorUseCase( + description = "Respond to a Slack message in a thread.", + keywords = {"slack", "respond", "reply", "thread"}, + configurations = { + @ProcessorConfiguration( + processorClassName = "org.apache.nifi.processors.standard.EvaluateJsonPath", + configuration = """ + Set "Destination" to "flowfile-attribute" + + Add a new property named "thread.ts" with a value of `$.threadTs` + Add a new property named "message.ts" with a value of `$.ts` + Add a new property named "channel.id" with a value of `$.channel` + Add a new property named "user.id" with a value of `$.user` + + Connect the "matched" Relationship to PublishSlack. + """ + ), + @ProcessorConfiguration( + processorClass = PublishSlack.class, + configuration = """ + Set "Access Token" to the value of your Slack OAuth Access Token. + Set "Channel" to `${'channel.id'}` + Set "Publish Strategy" to "Use 'Message Text' Property". + Set "Message Text" to the text that you would like to send as the response. If desired, you can reference the user of the original message by including the text `<@${'user.id'}>`. + For example: `Hey, <@${'user.id'}>, thanks for asking...` + Set "Include FlowFile Content as Attachment" to "false". + Set "Thread Timestamp" to `${'thread.ts':replaceEmpty( ${'message.ts'} )}` + """ + ) + } +) +public class PublishSlack extends AbstractProcessor { + + static final AllowableValue PUBLISH_STRATEGY_CONTENT_AS_MESSAGE = new AllowableValue("Send FlowFile Content as Message", "Send FlowFile Content as Message", + "The contents of the FlowFile will be sent as the message text."); + static final AllowableValue PUBLISH_STRATEGY_USE_PROPERTY = new AllowableValue("Use 'Message Text' Property", "Use 'Message Text' Property", + "The value of the Message Text Property will be sent as the message text."); + + static final PropertyDescriptor ACCESS_TOKEN = new PropertyDescriptor.Builder() + .name("Access Token") + .description("OAuth Access Token used for authenticating/authorizing the Slack request sent by NiFi. This may be either a User Token or a Bot Token. " + + "If a User Token, it must be granted the channels:write, groups:write, im:write, or mpim:write scope, depending on the type of conversation being used. " + + "If a Bot Token, it must be granted chat:write. Additionally, in order to upload FlowFile contents as an attachment, it must be granted files:write.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .sensitive(true) + .build(); + + protected PropertyDescriptor CHANNEL = new PropertyDescriptor.Builder() + .name("Channel") + .description("The name or identifier of the channel to send the message to. If using a channel name, it must be prefixed with the # character. " + + "For example, #general. This is valid only for public channels. Otherwise, the unique identifier of the channel to publish to must be " + + "provided.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .required(true) + .build(); + + protected PropertyDescriptor PUBLISH_STRATEGY = new PropertyDescriptor.Builder() + .name("Publish Strategy") + .description("Specifies how the Processor will send the message or file to Slack.") + .required(true) + .allowableValues(PUBLISH_STRATEGY_CONTENT_AS_MESSAGE, PUBLISH_STRATEGY_USE_PROPERTY) + .defaultValue(PUBLISH_STRATEGY_CONTENT_AS_MESSAGE.getValue()) + .build(); + + protected PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder() + .name("Character Set") + .description("Specifies the name of the Character Set used to encode the FlowFile contents.") + .required(true) + .defaultValue("UTF-8") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .dependsOn(PUBLISH_STRATEGY, PUBLISH_STRATEGY_CONTENT_AS_MESSAGE) + .build(); + + protected PropertyDescriptor MESSAGE_TEXT = new PropertyDescriptor.Builder() + .name("Message Text") + .description("The text of the message to send to Slack.") + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .required(true) + .addValidator(Validator.VALID) + .dependsOn(PUBLISH_STRATEGY, PUBLISH_STRATEGY_USE_PROPERTY) + .build(); + + protected PropertyDescriptor SEND_CONTENT_AS_ATTACHMENT = new PropertyDescriptor.Builder() + .name("Include FlowFile Content as Attachment") + .description("Specifies whether or not the contents of the FlowFile should be uploaded as an attachment to the Slack message.") + .allowableValues("true", "false") + .required(true) + .dependsOn(PUBLISH_STRATEGY, PUBLISH_STRATEGY_USE_PROPERTY) + .defaultValue("false") + .build(); + + protected PropertyDescriptor MAX_FILE_SIZE = new PropertyDescriptor.Builder() + .name("Max FlowFile Size") + .description("The maximum size of a FlowFile that can be sent to Slack. If any FlowFile exceeds this size, it will be routed to failure. " + + "This plays an important role because the entire contents of the file must be loaded into NiFi's heap in order to send the data " + + "to Slack.") + .required(true) + .dependsOn(SEND_CONTENT_AS_ATTACHMENT, "true") + .addValidator(StandardValidators.DATA_SIZE_VALIDATOR) + .defaultValue("1 MB") + .build(); + + protected PropertyDescriptor THREAD_TS = new PropertyDescriptor.Builder() + .name("Thread Timestamp") + .description("The Timestamp identifier for the thread that this message is to be a part of. If not specified, the message will be a top-level message instead of " + + "being in a thread.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .required(false) + .build(); + + + public static final Relationship REL_SUCCESS = new Relationship.Builder() + .name("success") + .description("FlowFiles are routed to success after being successfully sent to Slack") + .build(); + + public static final Relationship REL_RATE_LIMITED = new Relationship.Builder() + .name("rate limited") + .description("FlowFiles are routed to 'rate limited' if the Rate Limit has been exceeded") + .build(); + + public static final Relationship REL_TOO_LARGE = new Relationship.Builder() + .name("too large") + .description("FlowFiles are routed to 'too large' if they exceed Slack's limits or the Processor's configured size limit") + .build(); + + public static final Relationship REL_FAILURE = new Relationship.Builder() + .name("failure") + .description("FlowFiles are routed to 'failure' if unable to be sent to Slack for any other reason") + .build(); + + private final RateLimit rateLimit = new RateLimit(getLogger()); + + private volatile ChannelMapper channelMapper; + private volatile App slackApp; + private volatile MethodsClient client; + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return List.of(ACCESS_TOKEN, + CHANNEL, + PUBLISH_STRATEGY, + MESSAGE_TEXT, + CHARACTER_SET, + SEND_CONTENT_AS_ATTACHMENT, + MAX_FILE_SIZE, + THREAD_TS); Review Comment: Nope, will change that. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
