Merge "BUG-2383: rename BGPSessionImpl.sendMessage()"
[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / ChannelOutputLimiter.java
1 /*
2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.protocol.bgp.rib.impl;
9
10 import com.google.common.base.Preconditions;
11 import io.netty.channel.ChannelHandlerContext;
12 import io.netty.channel.ChannelInboundHandlerAdapter;
13 import javax.annotation.concurrent.ThreadSafe;
14 import org.opendaylight.yangtools.yang.binding.Notification;
15 import org.slf4j.Logger;
16 import org.slf4j.LoggerFactory;
17
18 /**
19  * A best-effort output limiter. It does not provide any fairness, and acts as a blocking gate-keeper
20  * for a sessions' channel.
21  */
22 @ThreadSafe
23 final class ChannelOutputLimiter extends ChannelInboundHandlerAdapter {
24     private static final Logger LOG = LoggerFactory.getLogger(ChannelOutputLimiter.class);
25     private final BGPSessionImpl session;
26     private volatile boolean blocked;
27
28     ChannelOutputLimiter(final BGPSessionImpl session) {
29         this.session = Preconditions.checkNotNull(session);
30     }
31
32     private void ensureWritable() {
33         if (blocked) {
34             LOG.trace("Blocked slow path tripped on session {}", session);
35             synchronized (this) {
36                 while (blocked) {
37                     try {
38                         LOG.debug("Waiting for session {} to become writable", session);
39                         this.wait();
40                     } catch (InterruptedException e) {
41                         throw new IllegalStateException("Interrupted while waiting for channel to come back", e);
42                     }
43                 }
44
45                 LOG.debug("Resuming write on session {}", session);
46             }
47         }
48     }
49
50     void write(final Notification msg) {
51         ensureWritable();
52         session.write(msg);
53     }
54
55     void writeAndFlush(final Notification msg) {
56         ensureWritable();
57         session.writeAndFlush(msg);
58     }
59
60     void flush() {
61         session.flush();
62     }
63
64     @Override
65     public void channelWritabilityChanged(final ChannelHandlerContext ctx) throws Exception {
66         final boolean w = ctx.channel().isWritable();
67
68         synchronized (this) {
69             blocked = !w;
70             LOG.debug("Writes on session {} {}", session, w ? "unblocked" : "blocked");
71
72             if (w) {
73                 this.notifyAll();
74             }
75         }
76
77         super.channelWritabilityChanged(ctx);
78     }
79
80     @Override
81     public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
82         synchronized (this) {
83             blocked = false;
84             this.notifyAll();
85         }
86
87         super.channelInactive(ctx);
88     }
89 }