| // 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 streamer; |
| |
| import org.apache.log4j.Logger; |
| |
| /** |
| * Link to transfer data in bounds of single thread (synchronized transfer). |
| * Must not be used to send data to elements served in different threads. |
| */ |
| public class SyncLink implements Link { |
| private static final Logger s_logger = Logger.getLogger(SyncLink.class); |
| |
| /** |
| * When null packet is pulled from source element, then make slight delay to |
| * avoid consuming of 100% of CPU in main loop in cases when link is pauses or |
| * source element cannot produce data right now. |
| */ |
| public static final long STANDARD_DELAY_FOR_EMPTY_PACKET = 10; // Milliseconds |
| |
| /** |
| * Delay for null packets in poll method when blocking is requested, in |
| * milliseconds. |
| */ |
| protected long delay = STANDARD_DELAY_FOR_EMPTY_PACKET; |
| |
| /** |
| * Set to true to print debugging messages. |
| */ |
| protected boolean verbose = System.getProperty("streamer.Link.debug", "false").equals("true");; |
| |
| /** |
| * ID of this link. |
| */ |
| protected String id = null; |
| |
| /** |
| * Buffer with data to hold because link is paused, on hold, or data is pushed |
| * back from output element. |
| */ |
| protected ByteBuffer cacheBuffer = null; |
| |
| /** |
| * Size of expected packet. Data must be hold in link until full packet will |
| * be read. |
| */ |
| protected int expectedPacketSize = 0; |
| |
| /** |
| * Number of packets and packet header transferred to element. |
| */ |
| protected int packetNumber = 0; |
| |
| /** |
| * Element to pull data from, when in pull mode. |
| */ |
| protected Element source = null; |
| |
| /** |
| * Element to send data to in both pull and push modes. |
| */ |
| protected Element sink = null; |
| |
| /** |
| * When in loop, indicates that loop must be stopped. |
| * |
| * @see run() |
| */ |
| private boolean shutdown = false; |
| |
| /** |
| * Indicates that event STREAM_START is passed over this link, so main loop |
| * can be started to pull data from source element. |
| */ |
| protected boolean started = false; |
| |
| /** |
| * Set to true to hold all data in link until it will be set to false again. |
| */ |
| protected boolean paused = false; |
| |
| /** |
| * Used by pull() method to hold all data in this link to avoid recursion when |
| * source element is asked to push new data to it outputs. |
| */ |
| protected boolean hold = false; |
| |
| /** |
| * Operate in pull mode instead of default push mode. In pull mode, link will |
| * ask it source element for new data. |
| */ |
| protected boolean pullMode = false; |
| |
| public SyncLink() { |
| } |
| |
| public SyncLink(String id) { |
| this.id = id; |
| } |
| |
| @Override |
| public void pushBack(ByteBuffer buf) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Buffer pushed back: " + buf + "."); |
| |
| if (cacheBuffer != null) { |
| ByteBuffer tmp = cacheBuffer.join(buf); |
| cacheBuffer.unref(); |
| cacheBuffer = tmp; |
| } else { |
| cacheBuffer = buf; |
| cacheBuffer.ref(); |
| } |
| |
| resetCursor(); |
| } |
| |
| private void resetCursor() { |
| // Reset cursor |
| cacheBuffer.cursor = 0; |
| } |
| |
| @Override |
| public void pushBack(ByteBuffer buf, int lengthOfFullPacket) { |
| pushBack(buf); |
| expectedPacketSize = lengthOfFullPacket; |
| } |
| |
| @Override |
| public String toString() { |
| return "SyncLink(" + ((id != null) ? id + ", " : "") + source + ":" + sink + ")"; |
| } |
| |
| /** |
| * Push data to sink. Call with null to push cached data. |
| */ |
| @Override |
| public void sendData(ByteBuffer buf) { |
| if (!hold && pullMode) |
| throw new RuntimeException("[" + this + "] ERROR: link is not in push mode."); |
| |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Incoming buffer: " + buf + "."); |
| |
| if (buf == null && cacheBuffer == null) |
| return; |
| |
| if (cacheBuffer != null && buf != null) { |
| // Join old data with fresh data |
| buf = cacheBuffer.join(buf); |
| cacheBuffer.unref(); |
| cacheBuffer = buf; |
| } |
| |
| // Store buffer in cache field to simplify following loop |
| if (buf != null) |
| cacheBuffer = buf; |
| |
| // When data pushed back and length of data is less than length of full |
| // packet, then feed data to sink element immediately |
| while (cacheBuffer != null) { |
| if (paused || hold) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Transfer is paused. Data in cache buffer: " + cacheBuffer + "."); |
| |
| // Wait until rest of packet will be read |
| return; |
| } |
| |
| if (expectedPacketSize > 0 && cacheBuffer.length < expectedPacketSize) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Transfer is suspended because available data is less than expected packet size. Expected packet size: " |
| + expectedPacketSize + ", data in cache buffer: " + cacheBuffer + "."); |
| |
| // Wait until rest of packet will be read |
| return; |
| } |
| |
| // Full packet or packet header is read, feed it to element |
| buf = cacheBuffer; |
| cacheBuffer = null; |
| expectedPacketSize = 0; |
| packetNumber++; |
| |
| if (sink == null) |
| throw new NullPointerException("[" + this + "] ERROR: Cannot send data to sink: sink is null. Data: " + buf + "."); |
| |
| sink.handleData(buf, this); |
| // cacheBuffer and expectedPacketSize can be changed at this time |
| } |
| |
| } |
| |
| @SuppressWarnings("incomplete-switch") |
| @Override |
| public void sendEvent(Event event, Direction direction) { |
| |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Event " + event + " is received."); |
| |
| // Shutdown main loop (if any) when STREAM_CLOSE event is received. |
| switch (event) { |
| case STREAM_START: { |
| if (!started) |
| started = true; |
| else |
| // Event already sent trough this link |
| return; |
| break; |
| } |
| case STREAM_CLOSE: { |
| if (!shutdown) |
| shutdown = true; |
| else |
| // Event already sent trough this link |
| return; |
| break; |
| } |
| case LINK_SWITCH_TO_PULL_MODE: { |
| setPullMode(); |
| break; |
| } |
| |
| } |
| |
| switch (direction) { |
| case IN: |
| source.handleEvent(event, direction); |
| break; |
| case OUT: |
| sink.handleEvent(event, direction); |
| break; |
| } |
| } |
| |
| @Override |
| public ByteBuffer pull(boolean block) { |
| if (!pullMode) |
| throw new RuntimeException("[" + this + "] ERROR: This link is not in pull mode."); |
| |
| if (hold) |
| throw new RuntimeException("[" + this + "] ERROR: This link is already on hold, waiting for data to be pulled in. Circular reference?"); |
| |
| if (paused) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Cannot pull, link is paused."); |
| |
| // Make slight delay in such case, to avoid consuming 100% of CPU |
| if (block) { |
| try { |
| Thread.sleep(100); |
| } catch (InterruptedException e) { |
| s_logger.info("[ignored] interrupted during pull", e); |
| } |
| } |
| |
| return null; |
| } |
| |
| // If data in cache can be sent immediately, |
| // then return it instead of asking for more data from source |
| if (cacheBuffer != null && (expectedPacketSize == 0 || (expectedPacketSize > 0 && cacheBuffer.length >= expectedPacketSize))) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Data pulled from cache buffer: " + cacheBuffer + "."); |
| |
| ByteBuffer tmp = cacheBuffer; |
| cacheBuffer = null; |
| return tmp; |
| } |
| |
| // Pause this link, so incoming data will not be sent to sink |
| // immediately, then ask source element for more data |
| try { |
| hold = true; |
| source.poll(block); |
| } finally { |
| hold = false; |
| } |
| |
| // Can return something only when data was stored in buffer |
| if (cacheBuffer != null && (expectedPacketSize == 0 || (expectedPacketSize > 0 && cacheBuffer.length >= expectedPacketSize))) { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Data pulled from source: " + cacheBuffer + "."); |
| |
| ByteBuffer tmp = cacheBuffer; |
| cacheBuffer = null; |
| return tmp; |
| } else { |
| return null; |
| } |
| } |
| |
| @Override |
| public Element setSink(Element sink) { |
| if (sink != null && this.sink != null) |
| throw new RuntimeException("[" + this + "] ERROR: This link sink element is already set. Link: " + this + ", new sink: " + sink + ", existing sink: " |
| + this.sink + "."); |
| |
| if (sink == null && cacheBuffer != null) |
| throw new RuntimeException("[" + this + "] ERROR: Cannot drop link: cache is not empty. Link: " + this + ", cache: " + cacheBuffer); |
| |
| this.sink = sink; |
| |
| return sink; |
| } |
| |
| @Override |
| public Element setSource(Element source) { |
| if (this.source != null && source != null) |
| throw new RuntimeException("[" + this + "] ERROR: This link source element is already set. Link: " + this + ", new source: " + source |
| + ", existing source: " + this.source + "."); |
| |
| this.source = source; |
| return source; |
| } |
| |
| @Override |
| public Element getSource() { |
| return source; |
| } |
| |
| @Override |
| public Element getSink() { |
| return sink; |
| } |
| |
| @Override |
| public void pause() { |
| if (paused) |
| throw new RuntimeException("[" + this + "] ERROR: Link is already paused."); |
| |
| paused = true; |
| |
| } |
| |
| @Override |
| public void resume() { |
| paused = false; |
| } |
| |
| /** |
| * Run pull loop to actively pull data from source and push it to sink. It |
| * must be only one pull loop per thread. |
| * |
| * Pull loop will start after event STREAM_START. This link and source element |
| * incoming links will be switched to pull mode before pull loop will be |
| * started using event LINK_SWITCH_TO_PULL_MODE. |
| */ |
| @Override |
| public void run() { |
| // Wait until even STREAM_START will arrive |
| while (!started) { |
| delay(); |
| } |
| |
| sendEvent(Event.LINK_SWITCH_TO_PULL_MODE, Direction.IN); |
| |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Starting pull loop."); |
| |
| // Pull source in loop |
| while (!shutdown) { |
| // Pull data from source element and send it to sink element |
| ByteBuffer data = pull(false); |
| if (data != null) |
| sink.handleData(data, this); |
| |
| if (!shutdown && data == null) { |
| // Make slight delay to avoid consuming of 100% of CPU |
| delay(); |
| } |
| } |
| |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Pull loop finished."); |
| |
| } |
| |
| protected void delay() { |
| try { |
| Thread.sleep(delay); |
| } catch (InterruptedException e) { |
| throw new RuntimeException("[" + this + "] ERROR: Interrupted in main loop.", e); |
| } |
| } |
| |
| @Override |
| public void setPullMode() { |
| if (verbose) |
| s_logger.debug("[" + this + "] INFO: Switching to PULL mode."); |
| |
| pullMode = true; |
| } |
| |
| @Override |
| public void drop() { |
| if (pullMode) |
| throw new RuntimeException("[" + this + "] ERROR: Cannot drop link in pull mode."); |
| |
| if (cacheBuffer != null) |
| throw new RuntimeException("[" + this + "] ERROR: Cannot drop link when cache contains data: " + cacheBuffer + "."); |
| |
| source.dropLink(this); |
| sink.dropLink(this); |
| } |
| } |
