/*
 * Copyright (c) 2015 Cisco Systems, Inc. and others.  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
 */
package org.opendaylight.protocol.bgp.rib.impl;

import com.google.common.base.Preconditions;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import javax.annotation.concurrent.ThreadSafe;
import org.opendaylight.yangtools.yang.binding.Notification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * A best-effort output limiter. It does not provide any fairness, and acts as a blocking gate-keeper
 * for a sessions' channel.
 */
@ThreadSafe
final class ChannelOutputLimiter extends ChannelInboundHandlerAdapter {
    private static final Logger LOG = LoggerFactory.getLogger(ChannelOutputLimiter.class);
    private final BGPSessionImpl session;
    private volatile boolean blocked;

    ChannelOutputLimiter(final BGPSessionImpl session) {
        this.session = Preconditions.checkNotNull(session);
    }

    private void ensureWritable() {
        if (this.blocked) {
            LOG.trace("Blocked slow path tripped on session {}", this.session);
            synchronized (this) {
                while (this.blocked) {
                    try {
                        LOG.debug("Waiting for session {} to become writable", this.session);
                        this.wait();
                    } catch (final InterruptedException e) {
                        throw new IllegalStateException("Interrupted while waiting for channel to come back", e);
                    }
                }

                LOG.debug("Resuming write on session {}", this.session);
            }
        }
    }

    void write(final Notification msg) {
        ensureWritable();
        this.session.write(msg);
    }

    ChannelFuture writeAndFlush(final Notification msg) {
        ensureWritable();
        return this.session.writeAndFlush(msg);
    }

    void flush() {
        this.session.flush();
    }

    @Override
    public void channelWritabilityChanged(final ChannelHandlerContext ctx) throws Exception {
        final boolean w = ctx.channel().isWritable();

        synchronized (this) {
            this.blocked = !w;
            LOG.debug("Writes on session {} {}", this.session, w ? "unblocked" : "blocked");

            if (w) {
                this.notifyAll();
            } else {
                flush();
            }
        }

        super.channelWritabilityChanged(ctx);
    }

    @Override
    public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
        synchronized (this) {
            this.blocked = false;
            this.notifyAll();
        }

        super.channelInactive(ctx);
    }
}