@Test public void testProcessMessage() throws Exception { when(deserializationSchema.isEndOfStream(any())).thenReturn(false).thenReturn(false); when(deserializationSchema.deserialize(pubSubMessage(FIRST_MESSAGE))).thenReturn(FIRST_MESSAGE); when(deserializationSchema.deserialize(pubSubMessage(SECOND_MESSAGE))).thenReturn(SECOND_MESSAGE); when(sourceContext.getCheckpointLock()).thenReturn("some object to lock on"); pubSubSource.open(null); pubSubSource.processMessage(sourceContext, asList(receivedMessage("firstAckId", pubSubMessage(FIRST_MESSAGE)), receivedMessage("secondAckId", pubSubMessage(SECOND_MESSAGE)))); verify(sourceContext, times(1)).getCheckpointLock(); verify(deserializationSchema, times(1)).isEndOfStream(FIRST_MESSAGE); verify(deserializationSchema, times(1)).deserialize(pubSubMessage(FIRST_MESSAGE)); verify(sourceContext, times(1)).collect(FIRST_MESSAGE); verify(acknowledgeOnCheckpoint, times(1)).addAcknowledgeId("firstAckId"); verify(deserializationSchema, times(1)).isEndOfStream(SECOND_MESSAGE); verify(deserializationSchema, times(1)).deserialize(pubSubMessage(SECOND_MESSAGE)); verify(sourceContext, times(1)).collect(SECOND_MESSAGE); verify(acknowledgeOnCheckpoint, times(1)).addAcknowledgeId("secondAckId"); }
void processMessage(SourceContext<OUT> sourceContext, List<ReceivedMessage> messages) throws Exception { synchronized (sourceContext.getCheckpointLock()) { for (ReceivedMessage message : messages) { acknowledgeOnCheckpoint.addAcknowledgeId(message.getAckId()); PubsubMessage pubsubMessage = message.getMessage(); OUT deserializedMessage = deserializationSchema.deserialize(pubsubMessage); if (deserializationSchema.isEndOfStream(deserializedMessage)) { cancel(); return; } sourceContext.collect(deserializedMessage); } } }
PubSubSource extends RichSourceFunction<OUT> implements ResultTypeQueryable<OUT>, ParallelSourceFunction<OUT>, CheckpointListener, ListCheckpointed<AcknowledgeIdsForCheckpoint<String>> { void processMessage(SourceContext<OUT> sourceContext, List<ReceivedMessage> messages) throws Exception { synchronized (sourceContext.getCheckpointLock()) { for (ReceivedMessage message : messages) { acknowledgeOnCheckpoint.addAcknowledgeId(message.getAckId()); PubsubMessage pubsubMessage = message.getMessage(); OUT deserializedMessage = deserializationSchema.deserialize(pubsubMessage); if (deserializationSchema.isEndOfStream(deserializedMessage)) { cancel(); return; } sourceContext.collect(deserializedMessage); } } } }
PubSubSource extends RichSourceFunction<OUT> implements ResultTypeQueryable<OUT>, ParallelSourceFunction<OUT>, CheckpointListener, ListCheckpointed<AcknowledgeIdsForCheckpoint<String>> { void processMessage(SourceContext<OUT> sourceContext, List<ReceivedMessage> messages) throws Exception { synchronized (sourceContext.getCheckpointLock()) { for (ReceivedMessage message : messages) { acknowledgeOnCheckpoint.addAcknowledgeId(message.getAckId()); PubsubMessage pubsubMessage = message.getMessage(); OUT deserializedMessage = deserializationSchema.deserialize(pubsubMessage); if (deserializationSchema.isEndOfStream(deserializedMessage)) { cancel(); return; } sourceContext.collect(deserializedMessage); } } } PubSubSource(PubSubDeserializationSchema<OUT> deserializationSchema,
				PubSubSubscriberFactory pubSubSubscriberFactory,
				Credentials credentials,
				int maxMessagesToAcknowledge,
				AcknowledgeOnCheckpointFactory acknowledgeOnCheckpointFactory); }
PubSubSource extends RichSourceFunction<OUT> implements ResultTypeQueryable<OUT>, ParallelSourceFunction<OUT>, CheckpointListener, ListCheckpointed<AcknowledgeIdsForCheckpoint<String>> { void processMessage(SourceContext<OUT> sourceContext, List<ReceivedMessage> messages) throws Exception { synchronized (sourceContext.getCheckpointLock()) { for (ReceivedMessage message : messages) { acknowledgeOnCheckpoint.addAcknowledgeId(message.getAckId()); PubsubMessage pubsubMessage = message.getMessage(); OUT deserializedMessage = deserializationSchema.deserialize(pubsubMessage); if (deserializationSchema.isEndOfStream(deserializedMessage)) { cancel(); return; } sourceContext.collect(deserializedMessage); } } } PubSubSource(PubSubDeserializationSchema<OUT> deserializationSchema,
				PubSubSubscriberFactory pubSubSubscriberFactory,
				Credentials credentials,
				int maxMessagesToAcknowledge,
				AcknowledgeOnCheckpointFactory acknowledgeOnCheckpointFactory); @Override void open(Configuration configuration); @Override void run(SourceContext<OUT> sourceContext); @Override void cancel(); @Override TypeInformation<OUT> getProducedType(); static DeserializationSchemaBuilder newBuilder(); @Override void notifyCheckpointComplete(long checkpointId); @Override List<AcknowledgeIdsForCheckpoint<String>> snapshotState(long checkpointId, long timestamp); @Override void restoreState(List<AcknowledgeIdsForCheckpoint<String>> state); }
PubSubSource extends RichSourceFunction<OUT> implements ResultTypeQueryable<OUT>, ParallelSourceFunction<OUT>, CheckpointListener, ListCheckpointed<AcknowledgeIdsForCheckpoint<String>> { void processMessage(SourceContext<OUT> sourceContext, List<ReceivedMessage> messages) throws Exception { synchronized (sourceContext.getCheckpointLock()) { for (ReceivedMessage message : messages) { acknowledgeOnCheckpoint.addAcknowledgeId(message.getAckId()); PubsubMessage pubsubMessage = message.getMessage(); OUT deserializedMessage = deserializationSchema.deserialize(pubsubMessage); if (deserializationSchema.isEndOfStream(deserializedMessage)) { cancel(); return; } sourceContext.collect(deserializedMessage); } } } PubSubSource(PubSubDeserializationSchema<OUT> deserializationSchema,
				PubSubSubscriberFactory pubSubSubscriberFactory,
				Credentials credentials,
				int maxMessagesToAcknowledge,
				AcknowledgeOnCheckpointFactory acknowledgeOnCheckpointFactory); @Override void open(Configuration configuration); @Override void run(SourceContext<OUT> sourceContext); @Override void cancel(); @Override TypeInformation<OUT> getProducedType(); static DeserializationSchemaBuilder newBuilder(); @Override void notifyCheckpointComplete(long checkpointId); @Override List<AcknowledgeIdsForCheckpoint<String>> snapshotState(long checkpointId, long timestamp); @Override void restoreState(List<AcknowledgeIdsForCheckpoint<String>> state); static final int NO_MAX_MESSAGES_TO_ACKNOWLEDGE_LIMIT; }