/**
 * Copyright (c) 2016 Bosch Software Innovations GmbH.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Bosch Software Innovations GmbH - initial creation
 */
package org.eclipse.hono.telemetry.impl;

import org.apache.qpid.proton.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Service;

/**
 * A telemetry adapter that simply logs and discards all messages.
 *
 */
@Service
@Profile("testing")
public final class MessageDiscardingTelemetryAdapter extends BaseTelemetryAdapter {

    private static final Logger LOG = LoggerFactory.getLogger(MessageDiscardingTelemetryAdapter.class);

    @Override
    public void processTelemetryData(final Message data, final String clientId) {
        LOG.debug("processing telemetry data [id: {}, to: {}]", data.getMessageId(), data.getAddress());
    }
}

@@ -424,16 +424,65 @@
 .impl;%0A%0A
+import java.util.HashMap;%0Aimport java.util.Map;%0A%0A
 import o
@@ -753,16 +753,172 @@
 ages.%0A *
+ %3Cp%3E%0A * It can be configured to pause a sender periodically for a certain amount of time in order%0A * to see if the sender implements flow control correctly.
 %0A */%0A@Se
@@ -1136,104 +1136,1030 @@
 s);%0A
-%0A    @Override%0A    public void processTelemetryData(final Message data, final String clientId) %7B
+    private final long pauseThreshold;%0A    private final long pausePeriod;%0A    private Map%3CString, LinkStatus%3E statusMap = new HashMap%3C%3E();%0A%0A    public MessageDiscardingTelemetryAdapter() %7B%0A        this(0, 0);%0A    %7D%0A%0A    /**%0A     * @param pauseThreshold the number of messages after which the sender is paused. If set to 0 (zero) the sender will%0A     *                       never be paused.%0A     * @param pausePeriod the number of milliseconds after which the sender is resumed.%0A     */%0A    public MessageDiscardingTelemetryAdapter(final long pauseThreshold, final long pausePeriod) %7B%0A        this.pauseThreshold = pauseThreshold;%0A        this.pausePeriod = pausePeriod;%0A    %7D%0A%0A    @Override%0A    public void processTelemetryData(final Message data, final String linkId) %7B%0A        LinkStatus status = statusMap.get(linkId);%0A        if (status == null) %7B%0A            LOG.debug(%22creating new link status object %5B%7B%7D%5D%22, linkId);%0A            status = new LinkStatus(linkId);%0A            statusMap.put(linkId, status);%0A        %7D
 %0A   
@@ -2212,16 +2212,34 @@
  %7B%7D, to:
+ %7B%7D, content-type:
  %7B%7D%5D%22, d
@@ -2275,18 +2275,1038 @@
 ddress()
-);
+,%0A                data.getContentType());%0A        status.onMsgReceived();%0A    %7D%0A%0A    private class LinkStatus %7B%0A        private long msgCount;%0A        private String linkId;%0A        private boolean suspended;%0A%0A        public LinkStatus(final String linkId) %7B%0A            this.linkId = linkId;%0A        %7D%0A%0A        public void onMsgReceived() %7B%0A            msgCount++;%0A            if (pauseThreshold %3E 0 && msgCount %25 pauseThreshold == 0) %7B%0A                pause();%0A            %7D%0A        %7D%0A%0A        public void pause() %7B%0A            LOG.debug(%22pausing link %5B%7B%7D%5D%22, linkId);%0A            this.suspended = true;%0A            sendFlowControlMessage(linkId, true);%0A            vertx.setTimer(pausePeriod, fired -%3E %7B%0A                vertx.runOnContext(run -%3E resume());%0A            %7D);%0A        %7D%0A%0A        private void resume() %7B%0A            if (suspended) %7B%0A                LOG.debug(%22resuming link %5B%7B%7D%5D%22, linkId);%0A                sendFlowControlMessage(linkId, false);%0A                this.suspended = false;%0A            %7D%0A        %7D
 %0A    %7D%0A%7D

Add option to periodically pause/resume the sender.