import org.msgpack.value.ArrayValue; //导入依赖的package包/类
public byte[] processPacket(Pair<SphinxHeader, byte[]> packet, BigInteger privk)
        throws SphinxException, IOException, CryptoException {
    SphinxProcessData sphinxProcessData = packer.decryptSphinxPacket(packet, privk);
    byte routingFlag = sphinxProcessData.routing[0];
    if (routingFlag == SphinxClient.DEST_FLAG) {
        Value value = packer.handleReceivedForward(sphinxProcessData.delta);
        ArrayValue outerTuple = value.asArrayValue();
        ArrayValue destination = outerTuple.get(0).asArrayValue();
        byte[] message = outerTuple.get(1).asBinaryValue().asByteArray();
        if (isDestinationSelf(destination)) {
            return message;
        } else {
            throw new RuntimeException("Received message not meant for us");
        }
    }
    throw new RuntimeException("Processed non-destination packet");
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
static void downloadPreviewRows(TDJobOperator j, String description, TaskState state, DurationInterval retryInterval)
{
    StringWriter out = new StringWriter();

    try {
        addCsvHeader(out, j.getResultColumnNames());

        List<ArrayValue> rows = downloadFirstResults(j, PREVIEW_ROWS, state, PREVIEW, retryInterval);
        if (rows.isEmpty()) {
            logger.info("preview of {}: (no results)", description, j.getJobId());
            return;
        }
        for (ArrayValue row : rows) {
            addCsvRow(out, row);
        }
    }
    catch (Exception ex) {
        logger.warn("Getting rows for preview failed. Ignoring this error.", ex);
        return;
    }

    logger.info("preview of {}:\r\n{}", description, out.toString());
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private boolean fetchJobResult(TDJobOperator job)
{
    Optional<ArrayValue> firstRow = pollingRetryExecutor(state, RESULT)
            .retryUnless(TDOperator::isDeterministicClientException)
            .withErrorMessage("Failed to download result of job '%s'", job.getJobId())
            .run(s -> job.getResult(
                    ite -> ite.hasNext()
                            ? Optional.of(ite.next())
                            : Optional.absent()));

    // There must be at least one row in the result for the wait condition to be fulfilled.
    if (!firstRow.isPresent()) {
        return false;
    }

    ArrayValue row = firstRow.get();
    if (row.size() < 1) {
        throw new TaskExecutionException("Got empty row in result of query");
    }

    Value firstCol = row.get(0);
    return isTruthy(firstCol);
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
@Override
public void messageReceived(IoSession session, Object message) throws Exception {
    logger.debug("Received {}: {}", message);
    // message is a HeapBuffer
    IoBuffer buffer = (IoBuffer) message;
    Unpacker unpacker = Unpacker.getUnpacker(buffer.array());
    ArrayValue values = unpacker.unpackValue().asArrayValue();
    // TODO: This might have to change to handle tagging of messages
    if (values.get(0).isBinaryValue()) {
        String type = values.get(0).asBinaryValue().asString();
        logger.info("Received {}", type);
        // Ignore dummy values
        // TODO: Wait what, Loopix sends DUMMY messages in plaintext?
        if (type.equals("DUMMY")) {
            return;
        } else {
            logger.warn("Received unknown message");
        }
    } else if (values.get(0).isArrayValue()) {
        SphinxHeader header = SphinxHeader.fromValue(values.get(0).asArrayValue());
        byte[] body = values.get(1).asBinaryValue().asByteArray();
        byte[] decryptedBody = cryptoClient.processPacket(new ImmutablePair<>(header, body), secret);
        // Assume we are sending/receiving text messages for now
        logger.info("Received: {}", new String(decryptedBody, Charset.forName("UTF-8")));
    } else {
        logger.warn("Received unknown message");
    }
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
public static SphinxHeader fromValue(ArrayValue values) {
    try {
        ECPoint alpha = Unpacker.unpackEcPoint(values.get(0));
        byte[] beta = values.get(1).asBinaryValue().asByteArray();
        byte[] gamma = values.get(2).asBinaryValue().asByteArray();
        return new SphinxHeader(alpha, beta, gamma);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
public static byte[] ecPointToByteArray(ECPoint point) throws IOException {
    MessageBufferPacker packer = MessagePack.newDefaultBufferPacker();
    byte[] pointData = point.getEncoded(false);
    ArrayValue values = new ImmutableArrayValueImpl(new Value[]{
            new ImmutableLongValueImpl(713),
            new ImmutableBinaryValueImpl(pointData)
    });
    packer.packValue(values);
    return packer.toByteArray();
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private Collection<?> deserializeCollection( ModuleDescriptor module, CollectionType collectionType,
                                             ArrayValue value ) throws IOException
{
    Collection<?> collection = collectionType.isSet() ? new LinkedHashSet( value.size() )
                                                      : new ArrayList( value.size() );
    for( Value element : value.list() )
    {
        collection.add( doDeserialize( module, collectionType.collectedType(), element ) );
    }
    return collection;
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private boolean fetchJobResult(int rows, TDJobOperator job)
{
    Optional<ArrayValue> firstRow = pollingRetryExecutor(state, RESULT)
            .retryUnless(TDOperator::isDeterministicClientException)
            .withErrorMessage("Failed to download result of job '%s'", job.getJobId())
            .withRetryInterval(retryInterval)
            .run(s -> job.getResult(
                    ite -> ite.hasNext()
                            ? Optional.of(ite.next())
                            : Optional.absent()));

    if (!firstRow.isPresent()) {
        throw new TaskExecutionException("Got unexpected empty result for count job: " + job.getJobId());
    }
    ArrayValue row = firstRow.get();
    if (row.size() != 1) {
        throw new TaskExecutionException("Got unexpected result row size for count job: " + row.size());
    }
    Value count = row.get(0);
    IntegerValue actualRows;
    try {
        actualRows = count.asIntegerValue();
    }
    catch (MessageTypeCastException e) {
        throw new TaskExecutionException("Got unexpected value type count job: " + count.getValueType());
    }

    return BigInteger.valueOf(rows).compareTo(actualRows.asBigInteger()) <= 0;
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private static void addCsvRow(Writer out, ArrayValue row)
    throws IOException
{
    boolean first = true;
    for (Value v : row) {
        if (first) { first = false; }
        else { out.write(DELIMITER_CHAR); }
        addCsvValue(out, v);
    }
    out.write("\r\n");
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
static Config buildStoreParams(ConfigFactory cf, TDJobOperator j, boolean storeLastResults, TaskState state, DurationInterval retryInterval)
{
    if (storeLastResults) {
        Config td = cf.create();

        List<ArrayValue> results = downloadFirstResults(j, 1, state, RESULT, retryInterval);
        Map<RawValue, Value> map = new LinkedHashMap<>();
        if (!results.isEmpty()) {
            ArrayValue row = results.get(0);
            List<String> columnNames = j.getResultColumnNames();
            for (int i = 0; i < Math.min(row.size(), columnNames.size()); i++) {
                map.put(ValueFactory.newString(columnNames.get(i)), row.get(i));
            }
        }
        MapValue lastResults = ValueFactory.newMap(map);
        try {
            td.set("last_results", new ObjectMapper().readTree(lastResults.toJson()));
        }
        catch (IOException ex) {
            throw Throwables.propagate(ex);
        }

        return cf.create().set("td", td);
    }
    else {
        return cf.create();
    }
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private static List<ArrayValue> downloadFirstResults(TDJobOperator j, int max, TaskState state, String stateKey, DurationInterval retryInterval)
{
    return pollingRetryExecutor(state, stateKey)
            .retryUnless(TDOperator::isDeterministicClientException)
            .withRetryInterval(retryInterval)
            .withErrorMessage("Failed to download result of job '%s'", j.getJobId())
            .run(s -> {
                try {
                    return j.getResult(ite -> {
                        List<ArrayValue> results = new ArrayList<>(max);
                        for (int i = 0; i < max; i++) {
                            if (ite.hasNext()) {
                                ArrayValue row = ite.next().asArrayValue();
                                results.add(row);
                            }
                            else {
                                break;
                            }
                        }
                        return results;
                    });
                }
                catch (TDClientHttpNotFoundException ex) {
                    // this happens if query is INSERT or CREATE. return empty results
                    return ImmutableList.of();
                }
            });
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private Config row(List<String> keys, ArrayValue values)
{
    Config config = configFactory.create();
    // TODO: fail on keys and values count mismatch?
    int n = Math.min(keys.size(), values.size());
    for (int i = 0; i < n; i++) {
        config.set(keys.get(i), value(values.get(i)));
    }
    return config;
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private boolean isDestinationSelf(ArrayValue dest) {
    return host.equals(dest.get(0).asStringValue().asString())
            && port == dest.get(1).asIntegerValue().asShort()
            && name.equals(dest.get(2).asStringValue().asString());
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private List<EventEntry> decodeMultiEventStream(final ArrayValue value) {
  return value.list().stream().map(this::decodeEntry).collect(Collectors.toList());
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private ArrayValue serializeIterable( Options options, Iterable<?> iterable )
{
    return serializeStream( options, StreamSupport.stream( iterable.spliterator(), false ) );
}
 

import org.msgpack.value.ArrayValue; //导入依赖的package包/类
private ArrayValue serializeStream( Options options, Stream<?> stream )
{
    return ValueFactory.newArray( stream.map( element -> doSerialize( options, element, false ) )
                                        .collect( toList() ) );
}