[
https://issues.apache.org/jira/browse/FLINK-2182?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14577135#comment-14577135
]
ASF GitHub Bot commented on FLINK-2182:
---------------------------------------
Github user mbalassi commented on a diff in the pull request:
https://github.com/apache/flink/pull/804#discussion_r31910945
--- Diff:
flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/functions/source/StatefulSequenceSource.java
---
@@ -0,0 +1,88 @@
+/**
+ * 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.api.functions.source;
+
+
+import org.apache.flink.streaming.api.checkpoint.Checkpointed;
+import org.apache.flink.streaming.runtime.tasks.StreamingRuntimeContext;
+
+/**
+ * A stateful streaming source that emits each number from a given
interval exactly once,
+ * possibly in parallel.
+ */
+public class StatefulSequenceSource extends
RichParallelSourceFunction<Long> implements Checkpointed<Long> {
+ private static final long serialVersionUID = 1L;
+
+ private final long start;
+ private final long end;
+
+ private long collected;
+
+ private volatile boolean isRunning = true;
+
+ /**
+ * Creates a source that emits all numbers from the given interval
exactly once.
+ *
+ * @param start Start of the range of numbers to emit.
+ * @param end End of the range of numbers to emit.
+ */
+ public StatefulSequenceSource(long start, long end) {
+ this.start = start;
+ this.end = end;
+ this.collected = 0;
+ }
+
+ @Override
+ public void run(SourceContext<Long> ctx) throws Exception {
+ final Object checkpointLock = ctx.getCheckpointLock();
+
+ StreamingRuntimeContext context = (StreamingRuntimeContext)
getRuntimeContext();
+
--- End diff --
We can get away with not casting this, nothing streaming specific is used.
> Add stateful Streaming Sequence Source
> --------------------------------------
>
> Key: FLINK-2182
> URL: https://issues.apache.org/jira/browse/FLINK-2182
> Project: Flink
> Issue Type: Improvement
> Components: eaming, Streaming
> Reporter: Aljoscha Krettek
> Assignee: Aljoscha Krettek
> Priority: Minor
>
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
