mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
Reuse code between internal and 'API' replication: remove ReplicatedInputKeyBytesData and ReplicatedInputValueBytesData
This commit is contained in:
leventov
@@ -443,8 +443,7 @@ public class ReplicatedChronicleMap<K, V, R> extends VanillaChronicleMap<K, V, R
|
||||
@Override
|
||||
public void readExternalEntry(@NotNull Bytes source) {
|
||||
try (CompiledReplicatedMapQueryContext<K, V, R> remoteOpContext = mapContext()) {
|
||||
remoteOpContext.initReplicationInput(source);
|
||||
remoteOpContext.processReplicatedEvent();
|
||||
remoteOpContext.processReplicatedEvent(source);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -28,8 +28,6 @@ import net.openhft.chronicle.hash.impl.stage.query.QuerySegmentStages;
|
||||
import net.openhft.chronicle.hash.impl.stage.query.SearchAllocatedChunks;
|
||||
import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputKeyBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputValueBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.WrappedValueBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceDataHolder;
|
||||
import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages;
|
||||
@@ -93,9 +91,6 @@ nested = {
|
||||
|
||||
ReplicatedMapAbsentDelegating.class,
|
||||
|
||||
ReplicatedInputKeyBytesData.class,
|
||||
ReplicatedInputValueBytesData.class,
|
||||
|
||||
DummyValueZeroData.class,
|
||||
|
||||
HashEntryChecksumStrategy.class,
|
||||
|
||||
-69
@@ -1,69 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 2015 higherfrequencytrading.com
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.map.impl.stage.data.bytes;
|
||||
|
||||
import net.openhft.chronicle.bytes.Bytes;
|
||||
import net.openhft.chronicle.bytes.RandomDataInput;
|
||||
import net.openhft.chronicle.hash.AbstractData;
|
||||
import net.openhft.chronicle.hash.impl.stage.hash.KeyBytesInterop;
|
||||
import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput;
|
||||
import net.openhft.sg.Stage;
|
||||
import net.openhft.sg.StageRef;
|
||||
import net.openhft.sg.Staged;
|
||||
|
||||
|
||||
@Staged
|
||||
public class ReplicatedInputKeyBytesData<K> extends AbstractData<K> {
|
||||
@StageRef ReplicatedInput<K, ?, ?> in;
|
||||
@StageRef KeyBytesInterop<K> ki;
|
||||
|
||||
@Stage("CachedBytesReplicatedInputKey") private K cachedBytesReplicatedInputKey;
|
||||
@Stage("CachedBytesReplicatedInputKey") private boolean cachedBytesReplicatedInputKeyRead =
|
||||
false;
|
||||
|
||||
private void initCachedBytesReplicatedInputKey() {
|
||||
cachedBytesReplicatedInputKey = getUsing(cachedBytesReplicatedInputKey);
|
||||
cachedBytesReplicatedInputKeyRead = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RandomDataInput bytes() {
|
||||
return in.replicatedInputBytes;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long offset() {
|
||||
return in.riKeyOffset;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long size() {
|
||||
return in.riKeySize;
|
||||
}
|
||||
|
||||
@Override
|
||||
public K get() {
|
||||
return cachedBytesReplicatedInputKey;
|
||||
}
|
||||
|
||||
@Override
|
||||
public K getUsing(K using) {
|
||||
Bytes inputBytes = in.replicatedInputBytes;
|
||||
inputBytes.readPosition(in.riKeyOffset);
|
||||
return ki.keyReader.read(inputBytes, size(), using);
|
||||
}
|
||||
}
|
||||
-69
@@ -1,69 +0,0 @@
|
||||
/*
|
||||
* Copyright (C) 2015 higherfrequencytrading.com
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package net.openhft.chronicle.map.impl.stage.data.bytes;
|
||||
|
||||
import net.openhft.chronicle.bytes.Bytes;
|
||||
import net.openhft.chronicle.bytes.RandomDataInput;
|
||||
import net.openhft.chronicle.hash.AbstractData;
|
||||
import net.openhft.chronicle.map.impl.stage.input.ReplicatedInput;
|
||||
import net.openhft.chronicle.map.impl.stage.map.ValueBytesInterop;
|
||||
import net.openhft.sg.Stage;
|
||||
import net.openhft.sg.StageRef;
|
||||
import net.openhft.sg.Staged;
|
||||
|
||||
|
||||
@Staged
|
||||
public class ReplicatedInputValueBytesData<V> extends AbstractData<V> {
|
||||
@StageRef ReplicatedInput<?, V, ?> in;
|
||||
@StageRef ValueBytesInterop<V> vi;
|
||||
|
||||
@Stage("CachedBytesReplicatedInputValue") private V cachedBytesReplicatedInputValue;
|
||||
@Stage("CachedBytesReplicatedInputValue") private boolean cachedBytesReplicatedInputValueRead =
|
||||
false;
|
||||
|
||||
private void initCachedBytesReplicatedInputValue() {
|
||||
cachedBytesReplicatedInputValue = getUsing(cachedBytesReplicatedInputValue);
|
||||
cachedBytesReplicatedInputValueRead = true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RandomDataInput bytes() {
|
||||
return in.replicatedInputBytes;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long offset() {
|
||||
return in.riValueOffset;
|
||||
}
|
||||
|
||||
@Override
|
||||
public long size() {
|
||||
return in.riValueSize;
|
||||
}
|
||||
|
||||
@Override
|
||||
public V get() {
|
||||
return cachedBytesReplicatedInputValue;
|
||||
}
|
||||
|
||||
@Override
|
||||
public V getUsing(V using) {
|
||||
Bytes inputBytes = in.replicatedInputBytes;
|
||||
inputBytes.readPosition(in.riValueOffset);
|
||||
return vi.valueReader.read(inputBytes, size(), using);
|
||||
}
|
||||
}
|
||||
@@ -19,16 +19,14 @@ package net.openhft.chronicle.map.impl.stage.input;
|
||||
import net.openhft.chronicle.bytes.Bytes;
|
||||
import net.openhft.chronicle.hash.Data;
|
||||
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
|
||||
import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
|
||||
import net.openhft.chronicle.hash.replication.RemoteOperationContext;
|
||||
import net.openhft.chronicle.map.Replica;
|
||||
import net.openhft.chronicle.map.impl.ReplicatedChronicleMapHolder;
|
||||
import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputKeyBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.data.bytes.ReplicatedInputValueBytesData;
|
||||
import net.openhft.chronicle.map.impl.stage.query.ReplicatedMapQuery;
|
||||
import net.openhft.chronicle.map.impl.stage.replication.ReplicationUpdate;
|
||||
import net.openhft.chronicle.map.replication.MapRemoteQueryContext;
|
||||
import net.openhft.sg.Stage;
|
||||
import net.openhft.sg.StageRef;
|
||||
import net.openhft.sg.Staged;
|
||||
|
||||
@@ -37,66 +35,44 @@ import net.openhft.sg.Staged;
|
||||
public abstract class ReplicatedInput<K, V, R> implements RemoteOperationContext<K>,
|
||||
MapRemoteQueryContext<K, V, R>, Replica.QueryContext<K, V> {
|
||||
|
||||
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
|
||||
@StageRef ReplicatedChronicleMapHolder<K, V, R> mh;
|
||||
@StageRef ReplicationUpdate<K> ru;
|
||||
@StageRef ReplicatedInputKeyBytesData<K> replicatedInputKeyBytesValue;
|
||||
@StageRef ReplicatedInputValueBytesData<V> replicatedInputValueBytesValue;
|
||||
@StageRef ReplicatedMapQuery<K, V, ?> q;
|
||||
@StageRef SegmentStages s;
|
||||
@StageRef DummyValueZeroData<V> dummyValue;
|
||||
|
||||
@Override
|
||||
public Data<V> dummyZeroValue() {
|
||||
checkOnEachPublicOperation.checkOnEachPublicOperation();
|
||||
return dummyValue;
|
||||
}
|
||||
|
||||
public Bytes replicatedInputBytes = null;
|
||||
public void processReplicatedEvent(Bytes replicatedInputBytes) {
|
||||
long bootstrapTimestamp = replicatedInputBytes.readLong();
|
||||
|
||||
public void initReplicatedInputBytes(Bytes replicatedInputBytes) {
|
||||
this.replicatedInputBytes = replicatedInputBytes;
|
||||
}
|
||||
long timestamp = replicatedInputBytes.readStopBit();
|
||||
byte identifier = replicatedInputBytes.readByte();
|
||||
ru.initReplicationUpdate(identifier, timestamp);
|
||||
|
||||
// ri for "replication input"
|
||||
@Stage("ReplicationInput") public long bootstrapTimestamp;
|
||||
@Stage("ReplicationInput") public long riKeySize = -1;
|
||||
@Stage("ReplicationInput") public long riValueSize;
|
||||
boolean isDeleted = replicatedInputBytes.readBoolean();
|
||||
long keySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes);
|
||||
long keyOffset = replicatedInputBytes.readPosition();
|
||||
|
||||
@Stage("ReplicationInput") public long riKeyOffset;
|
||||
@Stage("ReplicationInput") public long riValueOffset;
|
||||
mh.m().setLastModificationTime(identifier, bootstrapTimestamp);
|
||||
|
||||
@Stage("ReplicationInput") public long riTimestamp;
|
||||
@Stage("ReplicationInput") public byte riId;
|
||||
@Stage("ReplicationInput") public boolean isDeleted;
|
||||
|
||||
|
||||
public void initReplicationInput(Bytes replicatedInputBytes) {
|
||||
initReplicatedInputBytes(replicatedInputBytes);
|
||||
bootstrapTimestamp = replicatedInputBytes.readLong();
|
||||
riTimestamp = replicatedInputBytes.readStopBit();
|
||||
riId = replicatedInputBytes.readByte();
|
||||
ru.initReplicationUpdate(riId, riTimestamp);
|
||||
|
||||
isDeleted = replicatedInputBytes.readBoolean();
|
||||
riKeySize = mh.m().keySizeMarshaller.readSize(replicatedInputBytes);
|
||||
riKeyOffset = replicatedInputBytes.readPosition();
|
||||
if (!isDeleted) {
|
||||
replicatedInputBytes.readSkip(riKeySize);
|
||||
riValueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes);
|
||||
riValueOffset = replicatedInputBytes.readPosition();
|
||||
}
|
||||
}
|
||||
|
||||
public void processReplicatedEvent() {
|
||||
|
||||
mh.m().setLastModificationTime(riId, bootstrapTimestamp);
|
||||
|
||||
q.initInputKey(replicatedInputKeyBytesValue);
|
||||
q.initInputKey(q.getInputKeyBytesAsData(replicatedInputBytes, keyOffset, keySize));
|
||||
if (isDeleted) {
|
||||
s.innerUpdateLock.lock();
|
||||
mh.m().remoteOperations.remove(this);
|
||||
} else {
|
||||
replicatedInputBytes.readSkip(keySize);
|
||||
long valueSize = mh.m().valueSizeMarshaller.readSize(replicatedInputBytes);
|
||||
long valueOffset = replicatedInputBytes.readPosition();
|
||||
Data<V> value = q.wrapValueBytesAsData(replicatedInputBytes, valueOffset, valueSize);
|
||||
|
||||
s.innerWriteLock.lock();
|
||||
mh.m().remoteOperations.put(this, replicatedInputValueBytesValue);
|
||||
mh.m().remoteOperations.put(this, value);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user