LogPublisherRegistry.java

package io.jstach.rainbowgum;

import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

import io.jstach.rainbowgum.LogPublisher.PublisherFactory;
import io.jstach.rainbowgum.LogPublisher.PublisherProvider;
import io.jstach.rainbowgum.publisher.BatchSwapAsyncLogPublisher;
import io.jstach.rainbowgum.publisher.BatchSwapAsyncLogPublisherBuilder;
import io.jstach.rainbowgum.spi.RainbowGumServiceProvider;

/**
 * Registry of publishers. Rainbow Gum registers default publishers with the following
 * schemes:
 * <ul>
 * <li>{@value #SYNC_SCHEME} - default sync publisher</li>
 * <li>{@value #ASYNC_SCHEME} - default async publisher</li>
 * <li>{@value #BATCH_SWAP_SCHEME} - {@link BatchSwapAsyncLogPublisher}</li>
 * <li>{@value #DEFAULT_SCHEME} - by default this is the same as
 * {@link #SYNC_SCHEME}.</li>
 * </ul>
 * Plugins may override the above so that the above is automatically replaced with other
 * providers via the service loader.
 *
 * @see RainbowGumServiceProvider
 */
public sealed interface LogPublisherRegistry extends LogPublisher.PublisherProvider {

	/**
	 * Register a publisher provider by URI scheme.
	 * @param scheme uri scheme.
	 * @param publisherProvider provider.
	 */
	public void register(String scheme, LogPublisher.PublisherProvider publisherProvider);

	/**
	 * This is URI scheme for the async publisher used by the publisher builder. Call
	 * {@link #register(String, io.jstach.rainbowgum.LogPublisher.PublisherProvider)} with
	 * this scheme to replace the default async publisher.
	 * <p>
	 * A module registering a replacement must support the {@value #BUFFER_SIZE_NAME} URI
	 * query parameter. Both {@link PublisherFactory#ofAsync(Integer)} and
	 * {@link LogPublisher.AsyncLogPublisher.Builder#bufferSize(int)} pass the requested
	 * capacity through that parameter. The corresponding named property is
	 * {@value #BUFFER_SIZE_PROPERTY}.
	 */
	public static String ASYNC_SCHEME = "async";

	/**
	 * Explicit scheme for {@link BatchSwapAsyncLogPublisher}, independent of which
	 * implementation is registered as the default {@link #ASYNC_SCHEME} publisher.
	 */
	public static String BATCH_SWAP_SCHEME = "batchswap";

	/**
	 * This is the URI scheme for the sync publisher builder to find a sync publisher.
	 * Call {@link #register(String, io.jstach.rainbowgum.LogPublisher.PublisherProvider)}
	 * with this scheme to replace the default sync publisher.
	 */
	public static String SYNC_SCHEME = "sync";

	/**
	 * This is the URI scheme for the default publisher.
	 * {@link #register(String, io.jstach.rainbowgum.LogPublisher.PublisherProvider)} with
	 * this scheme to replace the default publisher.
	 */
	public static String DEFAULT_SCHEME = "default";

	/**
	 * Because publisher configurators through service provider discovery can replace the
	 * default schemes of {@link #ASYNC_SCHEME}, {@link #SYNC_SCHEME} all the builtin core
	 * providers are prefixed with this value so that they can still be used even if
	 * overriden.
	 */
	public static String BUILTIN_SCHEME_PREFIX = "core.";

	/**
	 * Key parameter for the URI scheme a publisher was selected with, without
	 * {@value #BUILTIN_SCHEME_PREFIX}.
	 */
	public static final String SCHEME = "scheme";

	/**
	 * Default async buffer size.
	 */
	public static final int ASYNC_BUFFER_SIZE = 1024;

	/**
	 * Buffer Size property name.
	 */
	public static final String BUFFER_SIZE_NAME = "bufferSize";

	/**
	 * Buffer Size Property for Async publishers.
	 */
	public static final String BUFFER_SIZE_PROPERTY = LogProperties.PUBLISHER_PREFIX + "{" + SCHEME + "}."
			+ BUFFER_SIZE_NAME;

	/**
	 * Default time in milliseconds an async publisher's close waits for queued events to
	 * be written.
	 */
	public static final int ASYNC_SHUTDOWN_TIMEOUT = 10_000;

	/**
	 * Shutdown timeout property name.
	 */
	public static final String SHUTDOWN_TIMEOUT_NAME = "shutdownTimeout";

	/**
	 * Milliseconds an async publisher's close waits for queued events to be written.
	 */
	public static final String SHUTDOWN_TIMEOUT_PROPERTY = LogProperties.PUBLISHER_PREFIX + "{" + SCHEME + "}."
			+ SHUTDOWN_TIMEOUT_NAME;

}

final class DefaultPublisherRegistry implements LogPublisherRegistry {

	private final ConcurrentHashMap<String, PublisherProvider> providers = new ConcurrentHashMap<String, LogPublisher.PublisherProvider>();

	@Override
	public PublisherFactory provide(LogProviderRef ref) {
		ref = DefaultLogProviderRef.normalize(ref);
		var uri = ref.uri();
		String scheme = uri.getScheme();
		if (scheme == null) {
			throw new IllegalArgumentException("URI is missing scheme. uri: " + uri);
		}
		var provider = providers.get(scheme);
		if (provider == null) {
			throw LogProviderRef.NotFoundException.of(ProviderModule.ComponentType.PUBLISHER, scheme, uri,
					providers.keySet());
		}
		return provider.provide(ref);
	}

	@Override
	public void register(String scheme, LogPublisher.PublisherProvider publisherProvider) {
		providers.put(scheme, publisherProvider);
	}

	/**
	 * Creates a publisher registry instance.
	 * @return publisher registry.
	 */
	static LogPublisherRegistry of() {
		var r = new DefaultPublisherRegistry();
		for (var p : DefaultPublisherProviders.values()) {
			r.register(p.scheme(), p);
			r.register(BUILTIN_SCHEME_PREFIX + p.scheme(), p);
		}
		return r;
	}

}

enum DefaultPublisherProviders implements LogPublisher.PublisherProvider {

	DEFAULT {
		@Override
		public String scheme() {
			return LogPublisherRegistry.DEFAULT_SCHEME;
		}

		@Override
		protected PublisherFactory provide(String name, LogProviderRef ref, LogProperties properties) {
			return (n, config, appenders) -> new DefaultSyncLogPublisher(appenders.asSingle());
		}
	},
	SYNC {
		@Override
		public String scheme() {
			return LogPublisherRegistry.SYNC_SCHEME;
		}

		@Override
		protected PublisherFactory provide(String name, LogProviderRef ref, LogProperties properties) {
			return (n, config, appenders) -> new DefaultSyncLogPublisher(appenders.asSingle());
		}

	},
	ASYNC {

		@Override
		public String scheme() {
			return LogPublisherRegistry.ASYNC_SCHEME;
		}

		@Override
		protected PublisherFactory provide(String name, LogProviderRef ref, LogProperties properties) {
			return BATCH_SWAP.provide(name, ref, properties);
		}
	},
	BATCH_SWAP {

		@Override
		public String scheme() {
			return LogPublisherRegistry.BATCH_SWAP_SCHEME;
		}

		@Override
		protected PublisherFactory provide(String name, LogProviderRef ref, LogProperties properties) {
			return new BatchSwapAsyncLogPublisherBuilder(name, schemeKey(ref)).fromProperties(properties, ref).build();
		}
	};

	@Override
	public PublisherFactory provide(LogProviderRef ref) {
		return (n, config, appenders) -> provide(n, ref, config.properties()).create(n, config, appenders);
	}

	public abstract String scheme();

	/*
	 * The scheme filling {scheme} in the publisher's keys: the one written in the URI, so
	 * async settings apply to whichever publisher is registered as async, without the
	 * builtin prefix.
	 */
	static String schemeKey(LogProviderRef ref) {
		String scheme = Objects.requireNonNull(ref.uri().getScheme());
		return LogProperties.removeKeyPrefix(scheme, LogPublisherRegistry.BUILTIN_SCHEME_PREFIX);
	}

	protected abstract PublisherFactory provide(String name, LogProviderRef ref, LogProperties properties);

}