This is wrapper around gRPC's managed channel builder that is spring-aware
188464
@Bean
	@ConditionalOnMissingBean(SpringAwareManagedChannelBuilder.class)
	public SpringAwareManagedChannelBuilder managedChannelBuilder(
			Optional<List<GrpcManagedChannelBuilderCustomizer>> customizers) {
		return new SpringAwareManagedChannelBuilder(customizers);
	}
12461
@Override
    public void channelActive(final ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
        if (selectBucketEnabled) {
            byte[] key = bucket.getBytes();
            short keyLength = (short) bucket.length();
            BinaryMemcacheRequest request = new DefaultBinaryMemcacheRequest(key);
            request.setOpcode(SELECT_BUCKET_OPCODE);
            request.setKeyLength(keyLength);
            request.setTotalBodyLength(keyLength);
            this.ctx.writeAndFlush(request);
        } else {
            //remove the handler if the feature is not enabled
            originalPromise.setSuccess();
            this.ctx.pipeline().remove(this);
            this.ctx.fireChannelActive();
        }
    }
869684
private void processWatchers(MethodSpec.Builder createdMethodBuilder) {
    createdMethodBuilder.addStatement("Proto p = __proto__");
    getMethodsWithAnnotation(component, Watch.class)
        .forEach(method -> processWatcher(createdMethodBuilder, method));
  }
1086894
function createReadStreamBackpressureManager(readableStream) {
  const manager = {
    waitPush: true,
    programmedPushs: [],
    programPush: function programPush(chunk, encoding, done) {
      // Store the current write
      manager.programmedPushs.push([chunk, encoding, done]);
      // Need to be async to avoid nested push attempts
      // Programm a push attempt
      setImmediate(manager.attemptPush);
      // Let's say we're ready for a read
      readableStream.emit('readable');
      readableStream.emit('drain');
    },
    attemptPush: function attemptPush() {
      let nextPush;

      if (manager.waitPush) {
        if (manager.programmedPushs.length) {
          nextPush = manager.programmedPushs.shift();
          manager.waitPush = readableStream.push(nextPush[0], nextPush[1]);
          nextPush[2]();
        }
      } else {
        setImmediate(() => {
          // Need to be async to avoid nested push attempts
          readableStream.emit('readable');
        });
      }
    },
  };

  // Patch the readable stream to manage reads
  readableStream._read = function streamFilterRestoreRead() {
    manager.waitPush = true;
    // Need to be async to avoid nested push attempts
    setImmediate(manager.attemptPush);
  };

  return manager;
}
478700
public static ManagedChannel newClientChannel(Transport transport, SocketAddress address,
        boolean tls, boolean testca, @Nullable String authorityOverride,
        int flowControlWindow, boolean directExecutor) {
    ManagedChannelBuilder<?> builder;
    if (transport == Transport.OK_HTTP) {
      builder = newOkHttpClientChannel(address, tls, testca);
    } else {
      try {
        builder = newNettyClientChannel(transport, address, tls, testca, flowControlWindow);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
    if (authorityOverride != null) {
      builder.overrideAuthority(authorityOverride);
    }

    if (directExecutor) {
      builder.directExecutor();
    } else {
      // TODO(carl-mastrangelo): This should not be necessary.  I don't know where this should be
      // put.  Move it somewhere else, or remove it if no longer necessary.
      // See: https://github.com/grpc/grpc-java/issues/2119
      builder.executor(getExecutor());
    }

    return builder.build();
  }
1045979
static ClientRequestResponseChannel clientChannel(
          final Configuration configuration,
          final ResponseChannelConsumer consumer,
          final Logger logger) throws Exception {
    return new ClientRequestResponseChannel(
            configuration.addressOfHost,
            consumer,
            configuration.readBufferPoolSize,
            configuration.readBufferSize,
            logger);
  }
151580
@Handler(channels = NetworkChannel.class)
    public void onAccepted(Accepted event, IOSubchannel netChannel) {
        new WebAppMsgChannel(event, netChannel);
    }
127366
private static <T> Flowable<Notification<T>> createWithParameters(Connection con, String sql,
            Flowable<List<Object>> parameterGroups, List<ParameterPlaceholder> parameterPlaceholders,
            BiFunction<NamedCallableStatement, List<Object>, Single<T>> single) {
        Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
        final Function<NamedCallableStatement, Flowable<Notification<T>>> flowableFactory = //
                stmt -> parameterGroups //
                        .flatMap(parameters -> single.apply(stmt, parameters).toFlowable()) //
                        .materialize() //
                        .doOnComplete(() -> Util.commit(stmt.stmt)) //
                        .doOnError(e -> Util.rollback(stmt.stmt));
        Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
        return Flowable.using(resourceFactory, flowableFactory, disposer, true);
    }
1221392
private ChannelInitializer<SocketChannel> getChannelInitializer(final PravegaNodeUri location,
                                                                    final FlowHandler handler) {
        final SslContext sslCtx = getSslContext();

        return new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline p = ch.pipeline();
                if (sslCtx != null) {
                    SslHandler sslHandler = sslCtx.newHandler(ch.alloc(), location.getEndpoint(), location.getPort());

                    if (clientConfig.isValidateHostName()) {
                        SSLEngine sslEngine = sslHandler.engine();
                        SSLParameters sslParameters = sslEngine.getSSLParameters();
                        sslParameters.setEndpointIdentificationAlgorithm("HTTPS");
                        sslEngine.setSSLParameters(sslParameters);
                    }
                    p.addLast(sslHandler);
                }
                p.addLast(
                        new ExceptionLoggingHandler(location.getEndpoint()),
                        new CommandEncoder(handler.getBatchSizeTracker()),
                        new LengthFieldBasedFrameDecoder(WireCommands.MAX_WIRECOMMAND_SIZE, 4, 4),
                        new CommandDecoder(),
                        handler);
            }
        };
    }
773006
private void reSubscribeByName(EventChannelStruct eventChannelStruct, String name) {

        //  Get the map and the callback structure for channel
        Hashtable<String, EventCallBackStruct>
                callBackMap = EventConsumer.getEventCallbackMap();
        EventCallBackStruct callbackStruct = null;
        Enumeration channelNames = callBackMap.keys();
        while (channelNames.hasMoreElements()) {
            String  key = (String) channelNames.nextElement();
            EventCallBackStruct eventStruct = callBackMap.get(key);
            if (eventStruct.channel_name.equals(name)) {
                callbackStruct = eventStruct;
            }
        }

        //  Get the callback structure
        if (callbackStruct!=null) {
            callbackStruct.consumer.reSubscribeByName(eventChannelStruct, name);
        }
    }
1335165
function (type, para) {

		// avoid flooding if SR_PUBLISH is sending streaming data
		SR.Log('[' + type + '] received');
		switch (type) {
			//
			// pubsub related
			//
			// when a new published message arrives 
			case 'SR_MSG':
			// handle server-published messages
			case 'SR_PUBLISH':
				if (onChannelMessages.hasOwnProperty(para.channel)) {
					if (typeof onChannelMessages[para.channel] !== 'function')
						SR.Error('channel [' + para.channel + '] handler is not a function');
					else
						onChannelMessages[para.channel](para.msg, para.channel);
				}
				else
					SR.Error('cannot find channel [' + para.channel + '] to publish'); 
				return true;
				
			// when a list of messages arrive (in array)
			case 'SR_MSGLIST':	
				var msg_list = para.msgs;			
				if (msg_list && msg_list.length > 0 && onChannelMessages.hasOwnProperty(para.channel)) {
					
					for (var i=0; i < msg_list.length; i++)
						onChannelMessages[para.channel](msg_list[i], para.channel);
				}				
				return true;
				
			// redirect to another webpage
			case 'SR_REDIRECT':
				window.location.href = para.url;
				return true;
			case "SR_NOTIFY" :
				SR.Warn('SR_NOTIFY para: ');
				SR.Warn(para);
				console.log(onChannelMessages);
				if (onChannelMessages.hasOwnProperty('notify'))
					onChannelMessages['notify'](para, 'notify');
				return true;
			//
			// login related
			//
			case "SR_LOGIN_RESPONSE":
			case "SR_LOGOUT_RESPONSE":
				replyLogin(para);
				return true;
			case "SR_MESSAGE":
				alert('SR_MESSAGE: ' + para.msg);
				return true;
			case "SR_WARNING":
				alert('SR_WARNING: ' + para.msg);
				return true;
			case "SR_ERROR":
				alert('SR_ERROR: ' + para.msg);
                return true;
				
			default:
				// check if custom handlers exist and can handle it
				if (responseHandlers.hasOwnProperty(type)) {
					
					var callbacks = responseHandlers[type];
					
					// extract rid if available 
					var rid = undefined;					
					if (para.hasOwnProperty('_rid') === true) {				
						rid = para['_rid'];
						delete para['_rid'];
					}
										
					if (rid) {
						callbacks[rid](para, type);	
						
						// remove callback once done
						if (rid !== 'keep') {
							delete callbacks[rid];
						}
					}
					// otherwise ALL registered callbacks will be called
					else {
						if (Object.keys(callbacks).length > 1)
							SR.Warn('[' + type + '] no rid in update, dispatching to first of ' +  Object.keys(callbacks).length + ' callbacks');
						
						// call the first in callbacks then remove it
						// so only one callback is called unless it's registered via the 'keep_callback' flag
						for (var key in callbacks) {
							callbacks[key](para, type);
							if (key !== 'keep') {
								delete callbacks[key];
								break;
							}
						}
					}

					return true;
				}
	
				// still un-handled
				console.error('onResponse: unrecongized type: ' + type);
				return false;
		}
	}
1262470
@SuppressWarnings({ "unchecked", "PMD.ShortVariable" })
    public <E extends EventBase<?>, C extends Channel> void forChannels(
            Class<C> type, BiConsumer<E, C> handler) {
        Arrays.stream(channels)
            .filter(c -> type.isAssignableFrom(c.getClass()))
            .forEach(c -> handler.accept((E) this, (C) c));
    }
93819
protected void configureLimits(final T builder, final String name) {
        final GrpcChannelProperties properties = getPropertiesFor(name);
        final Integer maxInboundMessageSize = properties.getMaxInboundMessageSize();
        if (maxInboundMessageSize != null) {
            builder.maxInboundMessageSize(maxInboundMessageSize);
        }
    }
1294566
public GrpcChannel build() throws AlluxioStatusException {
    ManagedChannel underlyingChannel =
        GrpcManagedChannelPool.INSTANCE().acquireManagedChannel(mChannelKey,
            mConfiguration.getMs(PropertyKey.NETWORK_CONNECTION_HEALTH_CHECK_TIMEOUT_MS),
            mConfiguration.getMs(PropertyKey.MASTER_GRPC_CHANNEL_SHUTDOWN_TIMEOUT));
    try {
      Channel clientChannel = underlyingChannel;

      if (mAuthenticateChannel) {
        // Create channel authenticator based on provided content.
        ChannelAuthenticator channelAuthenticator;
        if (mUseSubject) {
          channelAuthenticator = new ChannelAuthenticator(mParentSubject, mConfiguration);
        } else {
          channelAuthenticator =
              new ChannelAuthenticator(mUserName, mPassword, mImpersonationUser,
                  mConfiguration.getEnum(PropertyKey.SECURITY_AUTHENTICATION_TYPE, AuthType.class),
                  mConfiguration.getMs(PropertyKey.MASTER_GRPC_CHANNEL_AUTH_TIMEOUT));
        }
        // Get an authenticated wrapper channel over given managed channel.
        clientChannel = channelAuthenticator.authenticate(mServerAddress, underlyingChannel);
      }
      // Create the channel after authentication with the target.
      return new GrpcChannel(mChannelKey, clientChannel,
          mConfiguration.getMs(PropertyKey.MASTER_GRPC_CHANNEL_SHUTDOWN_TIMEOUT));
    } catch (Exception exc) {
      // Release the managed channel to the pool before throwing.
      GrpcManagedChannelPool.INSTANCE().releaseManagedChannel(mChannelKey,
          mConfiguration.getMs(PropertyKey.MASTER_GRPC_CHANNEL_SHUTDOWN_TIMEOUT));
      throw exc;
    }
  }
145340
private ControlFlushed createControlFlushed(SIBUuid12 stream)
    throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createControlFlushed", new Object[] {stream});

    ControlFlushed flushedMsg;

    // Create new message
    try
    {
      flushedMsg = cmf.createNewControlFlushed();
    }
    catch (MessageCreateFailedException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.createControlFlushed",
        "1:1730:1.241",
        this);
        
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      {
        SibTr.exception(tc, e);
        SibTr.exit(tc, "createControlFlushed", e);
      }
      
      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
          "1:1742:1.241",
          e });

      throw new SIResourceException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
            "1:1750:1.241",
            e },
          null),
        e);
    }

    // As we are using the Guaranteed Header - set all the attributes as
    // well as the ones we want.
    SIMPUtils.setGuaranteedDeliveryProperties(flushedMsg,
        messageProcessor.getMessagingEngineUuid(), 
        targetMEUuid,
        stream,
        null,
        destinationHandler.getUuid(),
        ProtocolType.UNICASTINPUT,
        GDConfig.PROTOCOL_VERSION); 
  
    flushedMsg.setPriority(SIMPConstants.CTRL_MSG_PRIORITY);
    flushedMsg.setReliability(Reliability.ASSURED_PERSISTENT);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createControlFlushed");

    return flushedMsg;
  }
475654
private Map getPublishedChannelParametersMap(String channelPublishId) throws PortalException {
        try {
            IPortletDefinitionRegistry registry =
                    PortletDefinitionRegistryLocator.getPortletDefinitionRegistry();
            IPortletDefinition def = registry.getPortletDefinition(channelPublishId);
            return def.getParametersAsUnmodifiableMap();
        } catch (Exception e) {
            throw new PortalException("Unable to acquire channel definition.", e);
        }
    }
649466
private <T> void manageSubscription(EventStream<T> stream, Consumer<T> consumer) {
        manageSubscription(stream.subscribe(consumer));
    }