Helper method to perform certain side effects when the channel is connected.
14386
private void channelActiveSideEffects(final ChannelHandlerContext ctx) {
        long interval = env().keepAliveInterval();
        if (env().continuousKeepAliveEnabled()) {
            continuousKeepAliveFuture = ctx.executor().scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    if (shouldSendKeepAlive()) {
                        createAndWriteKeepAlive(ctx);
                    }
                }
            }, interval, interval, TimeUnit.MILLISECONDS);
        }
    }
1231311
def set_channel(self, channel):
        
        self.channel = channel
        self.__record_progress(Status.CREATE_TREE)
1351661
def stop(self):
        
        self._stop.set()
        if self._channel is not None:
            self._channel.close()
472638
private void removeChannel(String name) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "CommsInboundChain removeChnanel", name);
        // Neither of the thrown exceptions are permanent failures: 
        // they usually indicate that we're the victim of a race.
        // If the CFW is also tearing down the chain at the same time 
        // (for example, the SSL feature was removed), then this could
        // fail.
        try {
            _cfw.removeChannel(name);
        } catch (ChannelException e) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                SibTr.debug(this, tc, "Error removing channel " + name, e);
            }
        } catch (ChainException e) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                SibTr.debug(this, tc, "Error removing channel " + name, e);
            }
        }
    }
909127
def _call_spy(self, *args, **kwargs):
      '''
      
      '''
      if self._spy_side_effect:
          if self._spy_side_effect_args or self._spy_side_effect_kwargs:
              self._spy_side_effect(
                  *self._spy_side_effect_args,
                  **self._spy_side_effect_kwargs)
          else:
              self._spy_side_effect(*args, **kwargs)

      return_value = self._stub.call_orig(*args, **kwargs)
      if self._spy_return:
          self._spy_return(return_value)

      return return_value
252517
public void unsubscribe(String channel) {
		ChannelSubscription subscribedChannel = subscribedChannels.get(channel);

		if (isUnsubscribeValid(channel, subscribedChannel)) {
			subscribedChannel.setSubscribeOnReconnected(false);
			unsubscribe(channel, true, subscribedChannel.isWithNotification());
		}

		//subscribedChannels.remove(channel);
	}
854291
private void fireOnGetBasic(ChannelEvent e) {
        List<EventListener> listeners = changes.getListenerList(AMQP);
        for (EventListener listener : listeners) {
            ChannelListener amqpListener = (ChannelListener)listener;
            amqpListener.onGetBasic(e);
        }
    }
1317942
def _recv_flow_ok(self, method_frame):
        '''
        
        '''
        self.channel._active = method_frame.args.read_bit()
        if self._flow_control_cb is not None:
            self._flow_control_cb()
1032048
def disconnect_signals(self):
        """"""
        # Disconnect the signals inside each Work.
        for work in self:
            work.disconnect_signals()

        # Disable callbacks.
        for cbk in self._callbacks:
            cbk.disable()
356703
def _handle_output(self, channel, output):
		
		augmented = Sentinel.augment_items(output, channel=channel, secret=False)
		for message in augmented:
			self.out(message.channel, message, not message.secret)
371193
public void onStateChange() {
        switch (dataChannel.state()) {
            case CONNECTING:
                break;

            case OPEN: {
                    if (null != callReference) {
                        RespokeCall call = callReference.get();
                        if (null != call) {
                            call.directConnectionDidOpen(this);
                        }
                    }

                new Handler(Looper.getMainLooper()).post(new Runnable() {
                    public void run() {
                        if (null != listenerReference) {
                            Listener listener = listenerReference.get();
                            if (null != listener) {
                                listener.onOpen(RespokeDirectConnection.this);
                            }
                        }
                    }
                });
                }
                break;

            case CLOSING:
                break;

            case CLOSED: {
                    if (null != callReference) {
                        RespokeCall call = callReference.get();
                        if (null != call) {
                            call.directConnectionDidClose(this);
                        }
                    }

                    new Handler(Looper.getMainLooper()).post(new Runnable() {
                        public void run() {
                            if (null != listenerReference) {
                                Listener listener = listenerReference.get();
                                if (null != listener) {
                                    listener.onClose(RespokeDirectConnection.this);
                                }
                            }
                        }
                    });
                }
                break;
        }
    }
939158
private void onBarrier(int channelIndex) throws IOException {
		if (!blockedChannels[channelIndex]) {
			blockedChannels[channelIndex] = true;

			numBarriersReceived++;

			if (LOG.isDebugEnabled()) {
				LOG.debug("{}: Received barrier from channel {}.",
					inputGate.getOwningTaskName(),
					channelIndex);
			}
		}
		else {
			throw new IOException("Stream corrupt: Repeated barrier for same checkpoint on input " + channelIndex);
		}
	}
458092
def _cleanup_channel(self, channel_id):
        
        with self.lock:
            if channel_id not in self._channels:
                return
            del self._channels[channel_id]
1243046
private void handleResponse(ByteBuf response, ChannelHandlerContext context) {
    NettyConnection connection = getConnection(context.channel());
    if (connection != null) {
      connection.handleResponse(response);
    }
  }
910152
def _cleanupConnections(senderkey, signal):
    """"""
    try:
        receivers = connections[senderkey][signal]
    except:
        pass
    else:
        if not receivers:
            # No more connected receivers. Therefore, remove the signal.
            try:
                signals = connections[senderkey]
            except KeyError:
                pass
            else:
                del signals[signal]
                if not signals:
                    # No more signal connections. Therefore, remove the sender.
                    _removeSender(senderkey)
85227
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Connected to: {}", getRemoteAddress());
        }
    }
728898
def drain_events(self, timeout=None):
        """"""
        chanmap = self.channels
        chanid, method_sig, args, content = self._wait_multiple(
            chanmap, None, timeout=timeout,
        )

        channel = chanmap[chanid]

        if (content and
                channel.auto_encode_decode and
                hasattr(content, 'content_encoding')):
            try:
                content.body = content.body.decode(content.content_encoding)
            except Exception:
                pass

        amqp_method = (self._method_override.get(method_sig) or
                       channel._METHOD_MAP.get(method_sig, None))

        if amqp_method is None:
            raise AMQPNotImplementedError(
                'Unknown AMQP method {0!r}'.format(method_sig))

        if content is None:
            return amqp_method(channel, args)
        else:
            return amqp_method(channel, args, content)