import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
public BigInteger unpackBigNumber() throws IOException {
    ImmutableValue value = this.unpackValue();
    if (!value.isExtensionValue())
        throw new RuntimeException("Expected extension value");

    ExtensionValue extValue = value.asExtensionValue();
    if (extValue.getType() != 0)
        throw new RuntimeException("Expected bignum value");

    byte[] data = extValue.getData();
    byte sign = data[0];
    if (sign != '+' && sign != '-')
        throw new RuntimeException("Sign was not pos or neg");

    BigInteger bn = new BigInteger(sign == '+' ? 1 : -1, Arrays.copyOfRange(data, 1, data.length));
    return bn;
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
@Override
public <T> T deserialize( ModuleDescriptor module, ValueType valueType, InputStream state )
{
    try( MessageUnpacker unpacker = MessagePack.newDefaultUnpacker( state ) )
    {
        if( !unpacker.hasNext() )
        {
            return null;
        }
        ImmutableValue value = unpacker.unpackValue();
        return doDeserialize( module, valueType, value );
    }
    catch( IOException ex )
    {
        throw new SerializationException( "Unable to deserialize " + valueType, ex );
    }
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
Complete(final ImmutableValue value) {
  this.value = value;
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
@Override
ImmutableValue value() {
  return value;
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
static DecodeResult complete(final ImmutableValue result) {
  return new Complete(result);
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
ConstantUnpacker(final int size, final Function<ByteBuffer, ImmutableValue> factory) {
  this.builder = ByteBuffer.allocate(size);
  this.factory = factory;
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
private MultipleUnpacker(final int size, final Function<ImmutableValue[], ImmutableValue> factory) {
  this.builder = new ImmutableValue[size];
  this.factory = factory;
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
public CheckPingResult decode(ImmutableValue value) {
  logger.debug("decoding ping {}", value);
  if (!value.isArrayValue()) {
    error("A ping message must be an array", value);
  }
  ImmutableArrayValue arrayValue = value.asArrayValue();
  if (arrayValue.size() != 6) {
    error("A ping message must have 6 elements", value);
  }
  String ping = decodeString(arrayValue.get(0));
  if (!ping.equals(("PING"))) {
    error("Invalid ping message", value);
  }
  String clientHostname = decodeString(arrayValue.get(1));
  String sharedKeySalt = decodeString(arrayValue.get(2)); // TODO Support both String and byte[]
  String sharedKeyHexDigest = decodeString(arrayValue.get(3));
  String username = decodeString(arrayValue.get(4));
  String passwordDigest = decodeString(arrayValue.get(5));

  if (node == null && !security.isAnonymousSourceAllowed()) {
    // FIXME add remote address to message
    String message = "Anonymous client disallowed.";
    logger.warn(message);
    return CheckPingResult.failure(message);
  }

  String sharedKey = null;
  try {
    sharedKey = node != null ? node.getSharedKey() : security.getSharedKey();
    MessageDigest md = MessageDigest.getInstance("SHA-512");
    md.update(sharedKeySalt.getBytes());
    md.update(clientHostname.getBytes());
    md.update(nonce);
    md.update(sharedKey.getBytes());
    String serverSideDigest = generateHexString(md.digest());
    if (!sharedKeyHexDigest.equals(serverSideDigest)) {
      // FIXME Add remote address to log
      logger.warn("Shared key mismatch: {}", clientHostname);
      return CheckPingResult.failure("Shared key mismatch");
    }

    if (security.isUserAuthEnabled()) {
      boolean userAuthenticationSucceeded = security.findAuthenticateUsers(node, username).stream().anyMatch(user -> {
        md.reset();
        md.update(userAuth);
        md.update(username.getBytes());
        md.update(user.getPassword().getBytes());
        String serverSidePasswordDigest = generateHexString(md.digest());
        return passwordDigest.equals(serverSidePasswordDigest);
      });
      if (!userAuthenticationSucceeded) {
        // FIXME Add remote address to log
        logger.info("Authentication failed: hostname={}, username={}", clientHostname, username);
        return CheckPingResult.failure("username/password mismatch");
      }
    }
  } catch (NoSuchAlgorithmException e) {
    error(e.getMessage(), value, e);
  }

  return CheckPingResult.success(sharedKeySalt, sharedKey);
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
/**
 * Decodes a request from clients.
 *
 * {@see https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1}
 *
 * {{{
 *   Connection ::= <<Request>>*
 *   Request ::= Message | Forward | PackedForward | nil
 *   Message ::= [ Tag, Time, Record, Option? ]
 *   Forward ::= [ Tag, MultiEventStream, Option? ]
 *   MultiEventStream ::= [ Event* ]
 *   PackedForward ::= [ Tag, MessagePackEventStream, Option? ]
 *   MessagePackEventStream ::= <<Event>>*
 *   Event ::= [ Time, Record ]
 *   Tag ::= string
 *   Time ::= integer | EventTime
 *   Record ::= object
 *   Option ::= object
 * }}}
 *
 * @param value msgpack value
 * @return {@code ForwardRequest} if {@code value} is a forward request,
 *         {@code Optional.empty()} if {@code value} is nil that is a future heartbeat request
 */
Optional<ForwardRequest> decode(final ImmutableValue value) {
  if (value.isNilValue()) return Optional.empty(); // reserved

  if (!value.isArrayValue()) {
    throw error("A request must be an array.", value);
  }
  final ImmutableArrayValue message = value.asArrayValue();

  if (message.size() < 2) {
    throw error("The size of array is too small.", value);
  }

  final Tag tag = decodeTag(message.get(0));

  final Value second = message.get(1);
  final List<EventEntry> entries;
  ForwardOption option = ForwardOption.empty();
  // TODO: CompressedPackedForward
  if (second.isArrayValue()) {
    entries = decodeMultiEventStream(second.asArrayValue());
    if (message.size() > 2) {
      option = decodeOption(message.get(2));
    }
  } else if (second.isStringValue() || second.isBinaryValue()) {
    entries = decodeMessagePackEventStream(second.asRawValue());
    if (message.size() > 2) {
      option = decodeOption(message.get(2));
    }
  } else {
    if (message.size() < 3) {
      throw error("The size of array is too small.", value);
    }
    final Instant time = decodeTime(second);
    final ImmutableMapValue record = decodeRecord(message.get(2));
    entries = Collections.singletonList(EventEntry.of(time, record));
    if (message.size() > 3) {
      option = decodeOption(message.get(3));
    }
  }

  return Optional.of(ForwardRequest.of(EventStream.of(tag, entries), option));
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
private void analyzeResult(String tag, ImmutableValue timestamp, Map<String, Object> data, long start, long end, boolean eventTime)
{
    Integer count = tagCounts.get(tag);
    if (count == null) {
        count = 0;
    }
    tagCounts.put(tag, count + 1);

    if (eventTime) {
        assertThat(timestamp.isExtensionValue(), is(true));
        ExtensionValue tsInEventTime = timestamp.asExtensionValue();
        assertThat(tsInEventTime.getType(), CoreMatchers.is((byte) 0x00));
        ByteBuffer secondsAndNanoSeconds = ByteBuffer.wrap(tsInEventTime.getData());
        int seconds = secondsAndNanoSeconds.getInt();
        int nanoSeconds = secondsAndNanoSeconds.getInt();
        assertTrue(start / 1000 <= seconds && seconds <= end / 1000);
        assertThat(nanoSeconds, is(999999999));
    }
    else {
        assertThat(timestamp.isIntegerValue(), is(true));
        long tsInEpochMilli = timestamp.asIntegerValue().asLong();
        assertTrue(start <= tsInEpochMilli && tsInEpochMilli<= end);
    }

    assertEquals(3, data.size());
    String name = (String) data.get("name");
    int age = (Integer) data.get("age");
    String comment = (String) data.get("comment");
    if (name.compareTo(minName) < 0) {
        minName = name;
    }
    if (name.compareTo(maxName) > 0) {
        maxName = name;
    }
    if (age < minAge) {
        minAge = age;
    }
    if (age > maxAge) {
        maxAge = age;
    }

    if (comment.equals("hello")) {
        // expected
    }
    else if (comment.equals(longStr)) {
        longCommentCount++;
    }
    else {
        assertTrue(false);
    }
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
@Override
public ImmutableValue immutableValue()
{
    return new ImmutableArrayValueImpl(list.toArray(new Value[list.size()]));
}
 

import org.msgpack.value.ImmutableValue; //导入依赖的package包/类
/**
 * @return the next {@code ImmutableValue}
 * @throws NoSuchElementException when no next value is found
 */
public ImmutableValue next() {
  return unpackedValues.remove();
}