[GitHub] [flink] senegalo commented on a change in pull request #12056: [FLINK-17502] [flink-connector-rabbitmq] RMQSource refactor

[GitBox <gi...@apache.org>]

senegalo commented on a change in pull request #12056:
URL: https://github.com/apache/flink/pull/12056#discussion_r453306517



##########
File path: flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQDeserializationSchema.java
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.flink.streaming.connectors.rabbitmq;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.util.Collector;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Envelope;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Interface for the set of methods required to parse an RMQ delivery.
+ * @param <T> The output type of the {@link RMQSource}
+ */
+public interface RMQDeserializationSchema<T> extends Serializable, ResultTypeQueryable<T> {
+	/**
+	 * This method takes all the RabbitMQ delivery information supplied by the client extract the data and pass it to the
+	 * collector.
+	 * NOTICE: The implementation of this method MUST call {@link RMQCollector#setMessageIdentifiers(String, long)} with
+	 * the correlation ID of the message if checkpointing and UseCorrelationID (in the RMQSource constructor) were enabled
+	 * the {@link RMQSource}.
+	 * @param envelope
+	 * @param properties
+	 * @param body
+	 * @throws IOException
+	 */
+	public  void processMessage(Envelope envelope, AMQP.BasicProperties properties, byte[] body, RMQCollector collector) throws IOException;
+
+	/**
+	 * Method to decide whether the element signals the end of the stream. If
+	 * true is returned the element won't be emitted.
+	 *
+	 * @param nextElement The element to test for the end-of-stream signal.
+	 * @return True, if the element signals end of stream, false otherwise.
+	 */
+	public boolean isEndOfStream(T nextElement);
+
+	/**
+	 * The {@link TypeInformation} for the deserialized T.
+	 * As an example the proper implementation of this method if T is a String is:
+	 * {@code return TypeExtractor.getForClass(String.class)}
+	 * @return TypeInformation
+	 */
+	public TypeInformation<T> getProducedType();
+
+
+	/**
+	 * Special collector for RMQ messages.
+	 * Captures the correlation ID and delivery tag also does the filtering logic for weather a message has been
+	 * processed or not.
+	 * @param <T>
+	 */
+	public interface RMQCollector<T> extends Collector<T> {
+		public void collect(List<T> records);
+
+		public void setMessageIdentifiers(String correlationId, long deliveryTag);
+	}

Review comment:
       I think i finally understand your point. 
   You're saying if there is a situation when 1 AMQP message received creates N messages to the collector 1:n there might be a situation where those N messages could have different corrlelation ids ? and there should be a way for them to be reset somehow ?
   
   but then:
   The suggestion above will not work since we create an instance of the collector every new message we get. So the reset is kinnda implied there.
   
   Also you can achieve the 1:N with multiple correlation IDs but just calling the `setMessageIdentifiers` multiple times.
   
   Let me know if got it right ... also that's my last working week then i have 2 weeks off .. 1 of them I'll be "safely" travelling :D the other one i intend to be creative and finish the 100 projects i am working on right now. One of the main topics is close this PR .. So I'll be like a car on the Autoban ... with No Speed Limits !!!! Vrooom Vroom 




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org