source.require(byteCount.toLong()) // Eagerly read the frame before firing client thread.
    source.read(buffer, byteCount.toLong())
    pushQueue.execute("$connectionName[$streamId] onData") {
      ignoreIoExceptions {
        val cancel = pushObserver.onData(streamId, buffer, byteCount, inFinished)
        if (cancel) writer.rstStream(streamId, ErrorCode.CANCEL)
        if (cancel || inFinished) {
          withLock {

      events.add(responseHeaders)
      notifyAll()
      return false
    }

    @Synchronized override fun onData(
      streamId: Int,
      source: BufferedSource,
      byteCount: Int,
      last: Boolean,
    ): Boolean {
      events.add(AssertionError("onData"))
      notifyAll()
      return false
    }

    @Synchronized override fun onReset(
      streamId: Int,

     * </p>
     *
     * @param inData
     *            The data to encode
     * @return The encoded data, or an empty string if the input is null or empty
     */
    public static String encode(final byte[] inData) {
        if (ArrayUtil.isEmpty(inData)) {
            return "";
        }
        return Base64.getEncoder().encodeToString(inData);
    }

    /**
     * Decodes data encoded in Base64.

                        logger.debug("member={}", memberOf);
                    }
                    String memberType = (String) memberOf.get("@odata.type");
                    if (memberType == null) {
                        logger.warn("@odata.type is null: {}", memberOf);
                        continue;
                    }
                    memberType = memberType.toLowerCase(Locale.ENGLISH);