--- /dev/null
+/*
+ * Copyright 2014-2021 Open Networking Foundation
+ * Copyright 2023 PANTHEON.tech, s.r.o.
+ *
+ * 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 io.atomix.utils.serializer;
+
+import static java.util.Objects.requireNonNull;
+
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.Registration;
+import com.esotericsoftware.kryo.Serializer;
+import com.esotericsoftware.kryo.io.ByteBufferInput;
+import com.esotericsoftware.kryo.io.ByteBufferOutput;
+import com.esotericsoftware.kryo.pool.KryoCallback;
+import com.esotericsoftware.kryo.pool.KryoFactory;
+import com.esotericsoftware.kryo.pool.KryoPool;
+import com.google.common.base.MoreObjects;
+import io.atomix.storage.journal.JournalSerdes;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Pool of Kryo instances, with classes pre-registered.
+ */
+final class KryoJournalSerdes implements JournalSerdes, KryoFactory, KryoPool {
+ /**
+ * Default buffer size used for serialization.
+ *
+ * @see #serialize(Object)
+ */
+ private static final int DEFAULT_BUFFER_SIZE = 4096;
+
+ /**
+ * Smallest ID free to use for user defined registrations.
+ */
+ private static final int INITIAL_ID = 16;
+
+ static final String NO_NAME = "(no name)";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(KryoJournalSerdes.class);
+
+ private final KryoPool kryoPool = new KryoPool.Builder(this).softReferences().build();
+
+ private final KryoOutputPool kryoOutputPool = new KryoOutputPool();
+ private final KryoInputPool kryoInputPool = new KryoInputPool();
+
+ private final List<RegisteredType> registeredTypes;
+ private final ClassLoader classLoader;
+ private final String friendlyName;
+
+ /**
+ * Creates a Kryo instance pool.
+ *
+ * @param registeredTypes types to register
+ * @param registrationRequired whether registration is required
+ * @param friendlyName friendly name for the namespace
+ */
+ KryoJournalSerdes(
+ final List<RegisteredType> registeredTypes,
+ final ClassLoader classLoader,
+ final String friendlyName) {
+ this.registeredTypes = List.copyOf(registeredTypes);
+ this.classLoader = requireNonNull(classLoader);
+ this.friendlyName = requireNonNull(friendlyName);
+
+ // Pre-populate with a single instance
+ release(create());
+ }
+
+ @Override
+ public byte[] serialize(final Object obj) {
+ return serialize(obj, DEFAULT_BUFFER_SIZE);
+ }
+
+ @Override
+ public byte[] serialize(final Object obj, final int bufferSize) {
+ return kryoOutputPool.run(output -> kryoPool.run(kryo -> {
+ kryo.writeClassAndObject(output, obj);
+ output.flush();
+ return output.getByteArrayOutputStream().toByteArray();
+ }), bufferSize);
+ }
+
+ @Override
+ public void serialize(final Object obj, final ByteBuffer buffer) {
+ ByteBufferOutput out = new ByteBufferOutput(buffer);
+ Kryo kryo = borrow();
+ try {
+ kryo.writeClassAndObject(out, obj);
+ out.flush();
+ } finally {
+ release(kryo);
+ }
+ }
+
+ @Override
+ public void serialize(final Object obj, final OutputStream stream) {
+ serialize(obj, stream, DEFAULT_BUFFER_SIZE);
+ }
+
+ @Override
+ public void serialize(final Object obj, final OutputStream stream, final int bufferSize) {
+ ByteBufferOutput out = new ByteBufferOutput(stream, bufferSize);
+ Kryo kryo = borrow();
+ try {
+ kryo.writeClassAndObject(out, obj);
+ out.flush();
+ } finally {
+ release(kryo);
+ }
+ }
+
+ @Override
+ public <T> T deserialize(final byte[] bytes) {
+ return kryoInputPool.run(input -> {
+ input.setInputStream(new ByteArrayInputStream(bytes));
+ return kryoPool.run(kryo -> {
+ @SuppressWarnings("unchecked")
+ T obj = (T) kryo.readClassAndObject(input);
+ return obj;
+ });
+ }, DEFAULT_BUFFER_SIZE);
+ }
+
+ @Override
+ public <T> T deserialize(final ByteBuffer buffer) {
+ ByteBufferInput in = new ByteBufferInput(buffer);
+ Kryo kryo = borrow();
+ try {
+ @SuppressWarnings("unchecked")
+ T obj = (T) kryo.readClassAndObject(in);
+ return obj;
+ } finally {
+ release(kryo);
+ }
+ }
+
+ @Override
+ public <T> T deserialize(final InputStream stream) {
+ return deserialize(stream, DEFAULT_BUFFER_SIZE);
+ }
+
+ @Override
+ public <T> T deserialize(final InputStream stream, final int bufferSize) {
+ ByteBufferInput in = new ByteBufferInput(stream, bufferSize);
+ Kryo kryo = borrow();
+ try {
+ @SuppressWarnings("unchecked")
+ T obj = (T) kryo.readClassAndObject(in);
+ return obj;
+ } finally {
+ release(kryo);
+ }
+ }
+
+ /**
+ * Creates a Kryo instance.
+ *
+ * @return Kryo instance
+ */
+ @Override
+ public Kryo create() {
+ LOGGER.trace("Creating Kryo instance for {}", this);
+ Kryo kryo = new Kryo();
+ kryo.setClassLoader(classLoader);
+ kryo.setRegistrationRequired(true);
+
+ // TODO rethink whether we want to use StdInstantiatorStrategy
+ kryo.setInstantiatorStrategy(
+ new Kryo.DefaultInstantiatorStrategy(new StdInstantiatorStrategy()));
+
+ int id = INITIAL_ID;
+ for (RegisteredType registeredType : registeredTypes) {
+ register(kryo, registeredType.types(), registeredType.serializer(), id++);
+ }
+ return kryo;
+ }
+
+ /**
+ * Register {@code type} and {@code serializer} to {@code kryo} instance.
+ *
+ * @param kryo Kryo instance
+ * @param types types to register
+ * @param serializer Specific serializer to register or null to use default.
+ * @param id type registration id to use
+ */
+ private void register(final Kryo kryo, final Class<?>[] types, final Serializer<?> serializer, final int id) {
+ Registration existing = kryo.getRegistration(id);
+ if (existing != null) {
+ boolean matches = false;
+ for (Class<?> type : types) {
+ if (existing.getType() == type) {
+ matches = true;
+ break;
+ }
+ }
+
+ if (!matches) {
+ LOGGER.error("{}: Failed to register {} as {}, {} was already registered.",
+ friendlyName, types, id, existing.getType());
+
+ throw new IllegalStateException(String.format(
+ "Failed to register %s as %s, %s was already registered.",
+ Arrays.toString(types), id, existing.getType()));
+ }
+ // falling through to register call for now.
+ // Consider skipping, if there's reasonable
+ // way to compare serializer equivalence.
+ }
+
+ for (Class<?> type : types) {
+ Registration r = null;
+ if (serializer == null) {
+ r = kryo.register(type, id);
+ } else if (type.isInterface()) {
+ kryo.addDefaultSerializer(type, serializer);
+ } else {
+ r = kryo.register(type, serializer, id);
+ }
+ if (r != null) {
+ if (r.getId() != id) {
+ LOGGER.debug("{}: {} already registered as {}. Skipping {}.",
+ friendlyName, r.getType(), r.getId(), id);
+ }
+ LOGGER.trace("{} registered as {}", r.getType(), r.getId());
+ }
+ }
+ }
+
+ @Override
+ public Kryo borrow() {
+ return kryoPool.borrow();
+ }
+
+ @Override
+ public void release(final Kryo kryo) {
+ kryoPool.release(kryo);
+ }
+
+ @Override
+ public <T> T run(final KryoCallback<T> callback) {
+ return kryoPool.run(callback);
+ }
+
+ @Override
+ public String toString() {
+ if (!NO_NAME.equals(friendlyName)) {
+ return MoreObjects.toStringHelper(getClass())
+ .omitNullValues()
+ .add("friendlyName", friendlyName)
+ // omit lengthy detail, when there's a name
+ .toString();
+ }
+ return MoreObjects.toStringHelper(getClass()).add("registeredTypes", registeredTypes).toString();
+ }
+}
Code Review / controller.git / blobdiff