import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer next() {
    ensureOpen();

    if (!hasNext())
        throw new NoSuchElementException();

    currentRow = keys.next();
    ByteBuffer currentKey = currentRow.key.getKey().duplicate();

    try {
        return StaticArrayBuffer.of(currentKey);
    } finally {
        lastSeenKey = currentKey;
    }
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public RowIterator(Closeable table, ResultScanner rows, byte[] columnFamilyBytes) {
    this.table = table;
    this.columnFamilyBytes = Arrays.copyOf(columnFamilyBytes, columnFamilyBytes.length);
    this.rows = Iterators.filter(rows.iterator(), new Predicate<Result>() {
        @Override
        public boolean apply(@Nullable Result result) {
            if (result == null)
                return false;

            try {
                StaticBuffer id = StaticArrayBuffer.of(result.getRow());
                id.getLong(0);
            } catch (NumberFormatException e) {
                return false;
            }

            return true;
        }
    });
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer next() {
    ensureOpen();

    if (!hasNext())
        throw new NoSuchElementException();

    currentRow = keys.next();
    ByteBuffer currentKey = currentRow.key.key.duplicate();

    try {
        return StaticArrayBuffer.of(currentKey);
    } finally {
        lastSeenKey = currentKey;
    }
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public RowIterator(HTableInterface table, ResultScanner rows) {
    this.table = table;
    this.rows = Iterators.filter(rows.iterator(), new Predicate<Result>() {
        @Override
        public boolean apply(@Nullable Result result) {
            if (result == null)
                return false;

            try {
                StaticBuffer id = StaticArrayBuffer.of(result.getRow());
                id.getLong(0);
            } catch (NumberFormatException e) {
                return false;
            }

            return true;
        }
    });
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer next() {
    ensureOpen();

    currentRow = rows.next();
    return StaticArrayBuffer.of(currentRow.getKey());
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Test
public void testFuzzMessagesSerial() throws Exception {
    final int maxLen = 1024 * 4;
    final int rounds = 32;

    StoringReader reader = new StoringReader(rounds);
    List<StaticBuffer> expected = new ArrayList<StaticBuffer>(rounds);

    Log l = manager.openLog("fuzz");
    l.registerReader(ReadMarker.fromNow(),reader);
    Random rand = new Random();
    for (int i = 0; i < rounds; i++) {
        //int len = rand.nextInt(maxLen + 1);
        int len = maxLen;
        if (0 == len)
            len = 1; // 0 would throw IllegalArgumentException
        byte[] raw = new byte[len];
        rand.nextBytes(raw);
        StaticBuffer sb = StaticArrayBuffer.of(raw);
        l.add(sb);
        expected.add(sb);
        Thread.sleep(50L);
    }
    reader.await(TIMEOUT_MS);
    assertEquals(rounds, reader.msgCount);
    assertEquals(expected, reader.msgs);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public void workerIterationStart(Configuration config, Configuration graphConfig, ScanMetrics metrics) {
    assertNotNull(config);
    metrics.incrementCustom(SETUP_COUNT);

    if (config.has(HEX_QUERIES)) {
        String queryStrings[] = config.get(HEX_QUERIES).split(":");
        List<SliceQuery> queries = new LinkedList<>();
        for (String qString : queryStrings) {
            String queryTokens[] = qString.split("/");
            StaticBuffer start = StaticArrayBuffer.of(Hex.hexToBytes(queryTokens[0]));
            StaticBuffer end = StaticArrayBuffer.of(Hex.hexToBytes(queryTokens[1]));
            SliceQuery query = new SliceQuery(start, end);
            int limit = Integer.valueOf(queryTokens[2]);
            if (0 <= limit) {
                query.setLimit(limit);
            }
            queries.add(query);
        }
        qs = queries;
    }

    if (config.has(KEY_FILTER_ID_MODULUS)) {
        final long mod = config.get(KEY_FILTER_ID_MODULUS);
        final long modVal;
        if (config.has(KEY_FILTER_ID_MODULAR_VALUE)) {
            modVal = config.get(KEY_FILTER_ID_MODULAR_VALUE);
        } else {
            modVal = 0;
        }
        keyFilter = k -> KeyValueStoreUtil.getID(k) % mod == modVal;
    }
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public static void loadValues(KeyColumnValueStore store, StoreTransaction tx, String[][] values, int shiftEveryNthRow,
                       int shiftSliceLength) throws BackendException {
    for (int i = 0; i < values.length; i++) {

        List<Entry> entries = new ArrayList<Entry>();
        for (int j = 0; j < values[i].length; j++) {
            StaticBuffer col;
            if (0 < shiftEveryNthRow && 0 == i/* +1 */ % shiftEveryNthRow) {
                ByteBuffer bb = ByteBuffer.allocate(shiftSliceLength + 9);
                for (int s = 0; s < shiftSliceLength; s++) {
                    bb.put((byte) -1);
                }
                bb.put(KeyValueStoreUtil.getBuffer(j + 1).asByteBuffer());
                bb.flip();
                col = StaticArrayBuffer.of(bb);

                // col = KeyValueStoreUtil.getBuffer(j + values[i].length +
                // 100);
            } else {
                col = KeyValueStoreUtil.getBuffer(j);
            }
            entries.add(StaticArrayEntry.of(col, KeyValueStoreUtil
                    .getBuffer(values[i][j])));
        }
        if (!entries.isEmpty()) {
            store.mutate(KeyValueStoreUtil.getBuffer(i), entries,
                    KeyColumnValueStore.NO_DELETIONS, tx);
        }
    }
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public static StaticBuffer stringToByteBuffer(String s) {
    byte[] b;
    try {
        b = s.getBytes("UTF-8");
    } catch (UnsupportedEncodingException e) {
        throw new RuntimeException(e);
    }
    ByteBuffer bb = ByteBuffer.allocate(b.length);
    bb.put(b);
    bb.flip();
    return StaticArrayBuffer.of(bb);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer getCurrentKey() throws IOException, InterruptedException {
    ImmutableBytesWritable ibw = (ImmutableBytesWritable)reader.getCurrentKey();
    StaticArrayBuffer sab = StaticArrayBuffer.of(ibw.copyBytes());
    return sab;
    // return StaticArrayBuffer.of(trb.getCurrentKey().copyBytes());
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public Entry next() {
    final Map.Entry<byte[], NavigableMap<Long, byte[]>> entry = iterator.next();
    byte[] col = entry.getKey();
    byte[] val = entry.getValue().lastEntry().getValue();
    return StaticArrayEntry.of(new StaticArrayBuffer(col), new StaticArrayBuffer(val));
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public TimestampRid fromLockColumn(StaticBuffer lockKey, TimestampProvider provider) {
    ReadBuffer r = lockKey.asReadBuffer();
    int len = r.length();
    long tsNS = r.getLong();
    len -= 8;
    byte[] curRid = new byte[len];
    for (int i = 0; r.hasRemaining(); i++) {
        curRid[i] = r.getByte();
    }
    StaticBuffer rid = new StaticArrayBuffer(curRid);
    Instant time = provider.getTime(tsNS);
    return new TimestampRid(time, rid);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer next() {
    ensureOpen();

    currentRow = rows.next();
    return StaticArrayBuffer.of(currentRow.getRow());
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
private KV completeNextKV() throws IOException {
    KV completedKV = null;
    boolean hasNext;
    do {
        hasNext = reader.nextKeyValue();

        if (!hasNext) {
            completedKV = incompleteKV;
            incompleteKV = null;
        } else {
            StaticArrayBuffer key = StaticArrayBuffer.of(reader.getCurrentKey());
            SortedMap<ByteBuffer, Cell> valueSortedMap = reader.getCurrentValue();
            List<Entry> entries = new ArrayList<>(valueSortedMap.size());
            for (Map.Entry<ByteBuffer, Cell> ent : valueSortedMap.entrySet()) {
                ByteBuffer col = ent.getKey();
                ByteBuffer val = ent.getValue().value();
                entries.add(StaticArrayEntry.of(StaticArrayBuffer.of(col), StaticArrayBuffer.of(val)));
            }

            if (null == incompleteKV) {
                // Initialization; this should happen just once in an instance's lifetime
                incompleteKV = new KV(key);
            } else if (!incompleteKV.key.equals(key)) {
                // The underlying Cassandra reader has just changed to a key we haven't seen yet
                // This implies that there will be no more entries for the prior key
                completedKV = incompleteKV;
                incompleteKV = new KV(key);
            }

            incompleteKV.addEntries(entries);
        }
        /* Loop ends when either
         * A) the cassandra reader ran out of data
         * or
         * B) the cassandra reader switched keys, thereby completing a KV */
    } while (hasNext && null == completedKV);

    return completedKV;
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Test
public void testFuzzMessages() throws Exception {
    final int maxLen = 1024 * 4;
    final int rounds = 32;

    StoringReader reader = new StoringReader(rounds);
    List<StaticBuffer> expected = new ArrayList<StaticBuffer>(rounds);

    Log l = manager.openLog("fuzz");
    l.registerReader(ReadMarker.fromNow(),reader);
    Random rand = new Random();
    for (int i = 0; i < rounds; i++) {
        //int len = rand.nextInt(maxLen + 1);
        int len = maxLen;
        if (0 == len)
            len = 1; // 0 would throw IllegalArgumentException
        byte[] raw = new byte[len];
        rand.nextBytes(raw);
        StaticBuffer sb = StaticArrayBuffer.of(raw);
        l.add(sb);
        expected.add(sb);
        Thread.sleep(50L);
    }
    reader.await(TIMEOUT_MS);
    assertEquals(rounds, reader.msgCount);
    assertEquals(expected, reader.msgs);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {

    // Consume columns on one row until end of iterator or a new key appears
    while (reader.nextKeyValue()) {
        StaticBuffer key = StaticArrayBuffer.of(reader.getCurrentKey().duplicate()); // TODO check and remove duplication
        Iterable<Entry> columns = new TitanCassandraHadoopGraph.CassandraMapIterable(reader.getCurrentValue());

        if (null == vb) {
            // Initialization on first iteration
            vb = graph.newVertexBuilder(configuration, key);
            Preconditions.checkArgument(vb.getKey().equals(key));
        } else if (!vb.getKey().equals(key)) {
            // Handle new row key
            vertex = vb.build();
            vertexQuery.filterRelationsOf(vertex);
            vb = graph.newVertexBuilder(configuration, key);
            // Vertex can be null if the system property representing lifecycle state (deletion) is set
            if (null != vertex)
                return true;
            // Continue iterating if it was null
        }

        // Add (column, value) pairs from current row
        vb.addEntries(columns);
    }

    // Iterator exhausted: check whether an unfinished vertex is ready for construction
    if (null != vb) {
        vertex = vb.build();
        vertexQuery.filterRelationsOf(vertex);
        vb = null;
        // Vertex can be null if the system property representing lifecycle state (deletion) is set
        return null != vertex;
    }

    return false;
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public Entry next() {
    final Map.Entry<ByteBuffer, Column> entry = iterator.next();
    ByteBuffer col = entry.getKey();
    ByteBuffer val = entry.getValue().value();
    return StaticArrayEntry.of(StaticArrayBuffer.of(col), StaticArrayBuffer.of(val));
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
public TimestampRid fromLockColumn(StaticBuffer lockKey) {
    ReadBuffer r = lockKey.asReadBuffer();
    int len = r.length();
    long tsNS = r.getLong();
    len -= 8;
    byte[] curRid = new byte[len];
    for (int i = 0; r.hasRemaining(); i++) {
        curRid[i] = r.getByte();
    }
    StaticBuffer rid = new StaticArrayBuffer(curRid);
    return new TimestampRid(tsNS, rid);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public StaticBuffer readObject(ObjectInput input) throws IOException,
        ClassNotFoundException {
    int len = input.readInt();
    byte[] raw = new byte[len];
    int actualLen = input.read(raw, 0, len);
    Preconditions.checkArgument(actualLen == len);
    return new StaticArrayBuffer(raw);
}
 

import com.thinkaurelius.titan.diskstorage.util.StaticArrayBuffer; //导入依赖的package包/类
@Override
public void replace(StaticBuffer key, StaticBuffer newValue, StaticBuffer oldValue, StoreTransaction txh) throws CacheUpdateException {
    byte[] rawKey = key.as(StaticArrayBuffer.ARRAY_FACTORY);
    byte[] rawNewValue = newValue.as(StaticArrayBuffer.ARRAY_FACTORY);

    // Hazelcast doesn't replace a value when old value was null
    // so we have to look and use putIfAbsent(new) if oldValue == null, otherwise use replace(old, new)
    if (oldValue == null) {
        if (cache.putIfAbsent(rawKey, newValue) != null)
            throw new CacheUpdateException(String.format(UPDATE_EXCEPTION_FORMAT, key, oldValue, newValue));
    } else if (!cache.replace(rawKey, oldValue, newValue)) {
        throw new CacheUpdateException(String.format(UPDATE_EXCEPTION_FORMAT, key, oldValue, newValue));
    }
}