class 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);

}