高性能内存消息和事件驱动库 Chronicle.7z

/* * Copyright 2011 Peter Lawrey * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package vanilla.java.chronicle.impl; import vanilla.java.chronicle.ByteString; import vanilla.java.chronicle.ByteStringAppender; import vanilla.java.chronicle.Chronicle; import vanilla.java.chronicle.Excerpt; import java.io.*; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.Arrays; import java.util.concurrent.atomic.AtomicBoolean; /** * @author peter.lawrey */ public abstract class AbstractExcerpt<C extends Chronicle> implements Excerpt<C> { protected final DirectChronicle chronicle; protected long index; protected long start = 0; protected long position = 0; protected int capacity = 0; protected long limit = 0; protected long startPosition; protected ByteBuffer buffer; private boolean forWrite = false; private static final byte[] MIN_VALUE_TEXT = ("" + Long.MIN_VALUE).getBytes(); private static final byte[] Infinity = "Infinity".getBytes(); private static final byte[] NaN = "NaN".getBytes(); private static final long MAX_VALUE_DIVIDE_5 = Long.MAX_VALUE / 5; private ExcerptInputStream inputStream = null; private ExcerptOutputStream outputStream = null; protected AbstractExcerpt(C chronicle) { this.chronicle = (DirectChronicle) chronicle; } @Override public C chronicle() { return (C) chronicle; } @Override public boolean index(long index) throws IndexOutOfBoundsException { readMemoryBarrier(); long endPosition = chronicle.getIndexData(index + 1); if (endPosition == 0) { capacity = 0; buffer = null; return false; } long startPosition = chronicle.getIndexData(index); capacity = (int) (endPosition - startPosition); index0(index, startPosition, endPosition); forWrite = false; // TODO Assumes the start of the record won't be all 0's // TODO Need to determine whether this is required as a safety check or not. return readLong(0) != 0L; } private boolean readMemoryBarrier() { return barrier.get(); } @Override public void startExcerpt(int capacity) { this.capacity = capacity; long startPosition = chronicle.startExcerpt(capacity); long endPosition = startPosition + capacity; index0(chronicle.size(), startPosition, endPosition); forWrite = true; } @Override public void finish() { if (position > limit) throw new IllegalStateException("Capacity allowed: " + capacity + " data read/written: " + (position - start)); if (forWrite) { final long endPosition = startPosition + (position - start); chronicle.setIndexData(index + 1, endPosition); chronicle.incrSize(); capacity = (int) (position - start); writeMemoryBarrier(); } } protected abstract void index0(long index, long startPosition, long endPosition); final AtomicBoolean barrier = new AtomicBoolean(); private void writeMemoryBarrier() { barrier.lazySet(true); } @Override public long index() { return index; } @Override public Excerpt<C> position(int position) { if (position < 0 || position > capacity()) throw new IndexOutOfBoundsException(); this.position = start + position; // start has to be added return this; } @Override public int position() { return (int) (position - start); } @Override public int capacity() { return (int) (limit - start); } @Override public int remaining() { return (int) (limit - position); } @Override public void readFully(byte[] b) { readFully(b, 0, b.length); } // RandomDataInput @Override public int skipBytes(int n) { int position = position(); int n2 = Math.min(n, capacity - position); position(position + n2); return n2; } @Override public void readFully(byte[] b, int off, int len) { while (len-- > 0) b[off++] = readByte(); } @Override public boolean readBoolean() { return readByte() != 0; } @Override public boolean readBoolean(int offset) { return readByte(offset) != 0; } @Override public int readUnsignedByte() { return readByte() & 0xFF; } @Override public int readUnsignedByte(int offset) { return readByte(offset) & 0xFF; } @Override public int readUnsignedShort() { return readShort() & 0xFFFF; } @Override public int readUnsignedShort(int offset) { return readShort(offset) & 0xFFFF; } @Override public String readLine() { StringBuilder input = new StringBuilder(); EOL: while (position() < capacity()) { int c = readUnsignedByte(); switch (c) { case '\n': break EOL; case '\r': int cur = position(); if (cur < capacity() && readByte(cur) == '\n') position(cur + 1); break EOL; default: input.append((char) c); break; } } return input.toString(); } @Override public String readUTF() { try { return DataInputStream.readUTF(this); } catch (IOException e) { throw new AssertionError(e); } } @Override public String readUTF(int offset) { long oldPosition = position; position = offset; try { return readUTF(); } finally { position = oldPosition; } } private static final byte BYTE_MIN_VALUE = Byte.MIN_VALUE; private static final byte BYTE_EXTENDED = Byte.MIN_VALUE + 1; private static final byte BYTE_MAX_VALUE = Byte.MIN_VALUE + 2; private static final short UBYTE_EXTENDED = 0xff; @Override public short readCompactShort() { byte b = readByte(); switch (b) { case BYTE_MIN_VALUE: return Short.MIN_VALUE; case BYTE_MAX_VALUE: return Short.MAX_VALUE; case BYTE_EXTENDED: return readShort(); default: return b; } } @Override public int readCompactUnsignedShort() { int b = readUnsignedByte(); if (b == UBYTE_EXTENDED) return readUnsignedShort(); return b; } @Override public int readInt24() { if (chronicle.byteOrder() == ByteOrder.BIG_ENDIAN) return (readUnsignedByte() << 24 + readUnsignedShort() << 8) >> 8; // extra shifting to get sign extension. return (readUnsignedByte() << 8 + readUnsignedShort() << 16) >> 8; } @Override public int readInt24(int offset) { if (chronicle.byteOrder() == ByteOrder.BIG_ENDIAN) return (readUnsignedByte(offset) << 24 + readUnsignedShort(offset + 1) << 8) >> 8; // extra shifting to get sign extension. return (readUnsignedByte(offset) << 8 + readUnsignedShort(offset + 1) << 16) >