spdis/voxy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rework to SectionRouter

Browse Source
This commit is contained in:
mcrcortex
2024-08-20 10:36:18 +10:00
parent 306956839a
commit fbdd65bc00
9 changed files with 235 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
src/main/java/me/cortex/voxy/client/core/rendering/RenderService.java
View File

@@ -2,9 +2,9 @@ package me.cortex.voxy.client.core.rendering;
import me.cortex.voxy.client.core.model.ModelBakerySubsystem;
import me.cortex.voxy.client.core.model.ModelStore;
import me.cortex.voxy.client.core.rendering.building.BuiltSection;
import me.cortex.voxy.client.core.rendering.building.RenderGenerationService;
import me.cortex.voxy.client.core.rendering.building.SectionUpdateRouter;
import me.cortex.voxy.client.core.rendering.building.SectionUpdate;
import me.cortex.voxy.client.core.rendering.hierachical2.HierarchicalNodeManager;
import me.cortex.voxy.client.core.rendering.hierachical2.HierarchicalOcclusionTraverser;
import me.cortex.voxy.client.core.rendering.section.AbstractSectionRenderer;
@@ -12,13 +12,14 @@ import me.cortex.voxy.client.core.rendering.section.IUsesMeshlets;
import me.cortex.voxy.client.core.rendering.section.MDICSectionRenderer;
import me.cortex.voxy.client.core.rendering.util.DownloadStream;
import me.cortex.voxy.client.core.rendering.util.UploadStream;
import me.cortex.voxy.common.util.MessageQueue;
import me.cortex.voxy.common.world.WorldEngine;
import me.cortex.voxy.common.thread.ServiceThreadPool;
import me.cortex.voxy.common.world.WorldSection;
import net.minecraft.client.render.Camera;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import static org.lwjgl.opengl.GL42.*;
@@ -37,7 +38,8 @@ public class RenderService<T extends AbstractSectionRenderer<J, ?>, J extends Vi
private final ModelBakerySubsystem modelService;
private final RenderGenerationService renderGen;
private final ConcurrentLinkedDeque<SectionUpdate> sectionUpdateQueue = new ConcurrentLinkedDeque<>();
private final MessageQueue<WorldSection> sectionUpdateQueue;
private final MessageQueue<BuiltSection> geometryUpdateQueue;
public RenderService(WorldEngine world, ServiceThreadPool serviceThreadPool) {
this.modelService = new ModelBakerySubsystem(world.getMapper());
@@ -47,23 +49,25 @@ public class RenderService<T extends AbstractSectionRenderer<J, ?>, J extends Vi
this.sectionRenderer = (T) createSectionRenderer(this.modelService.getStore(),1<<19, (1L<<30)-1024);
//Do something incredibly hacky, we dont need to keep the reference to this around, so just connect and discard
var positionFilterForwarder = new SectionUpdateRouter();
var router = new SectionUpdateRouter();
this.nodeManager = new HierarchicalNodeManager(1<<21, this.sectionRenderer.getGeometryManager(), positionFilterForwarder);
this.nodeManager = new HierarchicalNodeManager(1<<21, this.sectionRenderer.getGeometryManager(), router);
this.sectionUpdateQueue = new MessageQueue<>(section -> {
byte childExistence = section.getNonEmptyChildren();
section.release();//TODO: move this to another thread (probably a service job to free, this is because freeing can cause a DB save which should not happen on the render thread)
this.nodeManager.processChildChange(section.key, childExistence);
});
this.geometryUpdateQueue = new MessageQueue<>(this.nodeManager::processGeometryResult);
this.viewportSelector = new ViewportSelector<>(this.sectionRenderer::createViewport);
this.renderGen = new RenderGenerationService(world, this.modelService, serviceThreadPool, this.sectionUpdateQueue::add, this.sectionRenderer.getGeometryManager() instanceof IUsesMeshlets);
this.renderGen = new RenderGenerationService(world, this.modelService, serviceThreadPool, this.geometryUpdateQueue::push, this.sectionRenderer.getGeometryManager() instanceof IUsesMeshlets);
positionFilterForwarder.setCallbacks(this.renderGen::enqueueTask, section -> {
long time = SectionUpdate.getTime();
byte childExistence = section.getNonEmptyChildren();
this.sectionUpdateQueue.add(new SectionUpdate(section.key, time, null, childExistence));
});
router.setCallbacks(this.renderGen::enqueueTask, this.sectionUpdateQueue::push);
this.traversal = new HierarchicalOcclusionTraverser(this.nodeManager, 512);
world.setDirtyCallback(positionFilterForwarder::maybeForward);
world.setDirtyCallback(router::forward);
Arrays.stream(world.getMapper().getBiomeEntries()).forEach(this.modelService::addBiome);
world.getMapper().setBiomeCallback(this.modelService::addBiome);
@@ -107,10 +111,9 @@ public class RenderService<T extends AbstractSectionRenderer<J, ?>, J extends Vi
//TODO: Need to find a proper way to fix this (if there even is one)
if (true /* firstInvocationThisFrame */) {
DownloadStream.INSTANCE.tick();
//Process the build results here (this is done atomically/on the render thread)
while (!this.sectionUpdateQueue.isEmpty()) {
this.nodeManager.processResult(this.sectionUpdateQueue.poll());
}
this.sectionUpdateQueue.consume();
this.geometryUpdateQueue.consume();
}
UploadStream.INSTANCE.tick();
@@ -139,8 +142,7 @@ public class RenderService<T extends AbstractSectionRenderer<J, ?>, J extends Vi
this.sectionRenderer.free();
this.traversal.free();
//Release all the unprocessed built geometry
this.sectionUpdateQueue.forEach(update -> {if(update.geometry()!=null)update.geometry().free();});
this.sectionUpdateQueue.clear();
this.geometryUpdateQueue.clear(BuiltSection::free);
}
public Viewport<?> getViewport() {

14
src/main/java/me/cortex/voxy/client/core/rendering/building/RenderGenerationService.java
View File

@@ -23,13 +23,13 @@ public class RenderGenerationService {
private final WorldEngine world;
private final ModelBakerySubsystem modelBakery;
private final Consumer<SectionUpdate> resultConsumer;
private final Consumer<BuiltSection> resultConsumer;
private final boolean emitMeshlets;
private final ServiceSlice threads;
public RenderGenerationService(WorldEngine world, ModelBakerySubsystem modelBakery, ServiceThreadPool serviceThreadPool, Consumer<SectionUpdate> consumer, boolean emitMeshlets) {
public RenderGenerationService(WorldEngine world, ModelBakerySubsystem modelBakery, ServiceThreadPool serviceThreadPool, Consumer<BuiltSection> consumer, boolean emitMeshlets) {
this.emitMeshlets = emitMeshlets;
this.world = world;
this.modelBakery = modelBakery;
@@ -67,10 +67,10 @@ public class RenderGenerationService {
synchronized (this.taskQueue) {
task = this.taskQueue.removeFirst();
}
long time = SectionUpdate.getTime();
//long time = BuiltSection.getTime();
var section = task.sectionSupplier.get();
if (section == null) {
this.resultConsumer.accept(new SectionUpdate(task.position, time, BuiltSection.empty(task.position), (byte) 0));
this.resultConsumer.accept(BuiltSection.empty(task.position));
return;
}
section.assertNotFree();
@@ -103,11 +103,9 @@ public class RenderGenerationService {
}
}
byte childMask = section.getNonEmptyChildren();
section.release();
if (mesh != null) {
//Time is the time at the start of the update
this.resultConsumer.accept(new SectionUpdate(section.key, time, mesh, childMask));
if (mesh != null) {//If the mesh is null it means it didnt finish, so dont submit
this.resultConsumer.accept(mesh);
}
}

9
src/main/java/me/cortex/voxy/client/core/rendering/building/SectionUpdate.java
View File

@@ -1,9 +0,0 @@
package me.cortex.voxy.client.core.rendering.building;
import org.jetbrains.annotations.Nullable;
public record SectionUpdate(long position, long buildTime, @Nullable BuiltSection geometry, byte childExistence) {
public static long getTime() {
return System.currentTimeMillis();
}
}

75
src/main/java/me/cortex/voxy/client/core/rendering/building/SectionUpdateRouter.java
View File

@@ -1,77 +1,90 @@
package me.cortex.voxy.client.core.rendering.building;
import it.unimi.dsi.fastutil.longs.Long2ByteOpenHashMap;
import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
import me.cortex.voxy.common.world.WorldEngine;
import me.cortex.voxy.common.world.WorldSection;
import java.util.function.LongConsumer;
import static me.cortex.voxy.common.world.WorldEngine.UPDATE_TYPE_BLOCK_BIT;
public class SectionUpdateRouter {
private static final int SLICES = 1<<2;
private static final int SLICES = 1<<3;
public interface IChildUpdate {void accept(WorldSection section);}
private final LongOpenHashSet[] slices = new LongOpenHashSet[SLICES];
private final Long2ByteOpenHashMap[] slices = new Long2ByteOpenHashMap[SLICES];
{
for (int i = 0; i < this.slices.length; i++) {
this.slices[i] = new LongOpenHashSet();
this.slices[i] = new Long2ByteOpenHashMap();
}
}
private LongConsumer renderForwardTo;
private LongConsumer renderMeshGen;
private IChildUpdate childUpdateCallback;
public void setCallbacks(LongConsumer forwardTo, IChildUpdate childUpdateCallback) {
if (this.renderForwardTo != null) {
public void setCallbacks(LongConsumer renderMeshGen, IChildUpdate childUpdateCallback) {
if (this.renderMeshGen != null) {
throw new IllegalStateException();
}
this.renderForwardTo = forwardTo;
this.renderMeshGen = renderMeshGen;
this.childUpdateCallback = childUpdateCallback;
}
public boolean watch(int lvl, int x, int y, int z) {
return this.watch(WorldEngine.getWorldSectionId(lvl, x, y, z));
public boolean watch(int lvl, int x, int y, int z, int types) {
return this.watch(WorldEngine.getWorldSectionId(lvl, x, y, z), types);
}
public boolean watch(long position) {
public boolean watch(long position, int types) {
var set = this.slices[getSliceIndex(position)];
boolean added;
byte delta = 0;
synchronized (set) {
added = set.add(position);
byte current = 0;
if (set.containsKey(position)) {
current = set.get(position);
}
delta = (byte) ((current&types)^types);
current |= (byte) types;
set.put(position, current);
}
if (added) {
if ((delta&UPDATE_TYPE_BLOCK_BIT)!=0) {
//If we added it, immediately invoke for an update
this.renderForwardTo.accept(position);
this.renderMeshGen.accept(position);
}
return added;
return delta!=0;
}
public boolean unwatch(int lvl, int x, int y, int z) {
return this.unwatch(WorldEngine.getWorldSectionId(lvl, x, y, z));
public boolean unwatch(int lvl, int x, int y, int z, int types) {
return this.unwatch(WorldEngine.getWorldSectionId(lvl, x, y, z), types);
}
public boolean unwatch(long position) {
public boolean unwatch(long position, int types) {
var set = this.slices[getSliceIndex(position)];
synchronized (set) {
return set.remove(position);
byte current = set.get(position);
byte delta = (byte) (current&types);
current &= (byte) ~types;
if (current == 0) {
set.remove(position);
}
return delta!=0;
}
}
public void maybeForward(WorldSection section, int type) {
public void forward(WorldSection section, int type) {
final long position = section.key;
var set = this.slices[getSliceIndex(position)];
boolean contains;
byte types = 0;
synchronized (set) {
contains = set.contains(position);
types = set.getOrDefault(position, (byte)0);
}
if (contains) {
if (type == 3) {//If its both, propagate to the render service
this.renderForwardTo.accept(position);
} else {
if (type == 2) {//If its only a existance update
this.childUpdateCallback.accept(section);
} else {//If its only a geometry update
this.renderForwardTo.accept(position);
}
if (types!=0) {
if ((type&WorldEngine.UPDATE_TYPE_CHILD_EXISTENCE_BIT)!=0) {
this.childUpdateCallback.accept(section);
}
if ((type& UPDATE_TYPE_BLOCK_BIT)!=0) {
this.renderMeshGen.accept(section.key);
}
}
}

166
src/main/java/me/cortex/voxy/client/core/rendering/hierachical2/HierarchicalNodeManager.java
View File

@@ -5,10 +5,10 @@ import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
import it.unimi.dsi.fastutil.longs.Long2IntOpenHashMap;
import me.cortex.voxy.client.core.rendering.building.BuiltSection;
import me.cortex.voxy.client.core.rendering.building.SectionUpdateRouter;
import me.cortex.voxy.client.core.rendering.building.SectionUpdate;
import me.cortex.voxy.client.core.rendering.section.AbstractSectionGeometryManager;
import me.cortex.voxy.client.core.util.ExpandingObjectAllocationList;
import me.cortex.voxy.common.world.WorldEngine;
import me.cortex.voxy.common.world.WorldSection;
import me.jellysquid.mods.sodium.client.util.MathUtil;
import org.lwjgl.system.MemoryUtil;
@@ -56,7 +56,7 @@ public class HierarchicalNodeManager {
throw new IllegalArgumentException("Position already in node set: " + WorldEngine.pprintPos(position));
}
this.activeSectionMap.put(position, SENTINAL_TOP_NODE_INFLIGHT);
this.updateRouter.watch(position);
this.updateRouter.watch(position, WorldEngine.UPDATE_FLAGS);
}
public void removeTopLevelNode(long position) {
@@ -65,64 +65,6 @@ public class HierarchicalNodeManager {
}
}
public void processRequestQueue(int count, long ptr) {
for (int requestIndex = 0; requestIndex < count; requestIndex++) {
int op = MemoryUtil.memGetInt(ptr + (requestIndex * 4L));
this.processRequest(op);
}
}
private void processRequest(int op) {
int node = op& NODE_ID_MSK;
if (!this.nodeData.nodeExists(node)) {
throw new IllegalStateException("Tried processing a node that doesnt exist: " + node);
}
if (this.nodeData.isNodeRequestInFlight(node)) {
throw new IllegalStateException("Tried processing a node that already has a request in flight: " + node + " pos: " + WorldEngine.pprintPos(this.nodeData.nodePosition(node)));
}
this.nodeData.markRequestInFlight(node);
//2 branches, either its a leaf node -> emit a leaf request
// or the nodes geometry must be empty (i.e. culled from the graph/tree) so add to tracker and watch
if (this.nodeData.isLeafNode(node)) {
this.makeLeafRequest(node, this.nodeData.getNodeChildExistence(node));
} else {
//Verify that the node section is not in the section store. if it is then it is a state desynchonization
// Note that a section can be "empty" but some of its children might not be
}
}
private void makeLeafRequest(int node, byte childExistence) {
long pos = this.nodeData.nodePosition(node);
//Enqueue a leaf expansion request
var request = new NodeChildRequest(pos);
int requestId = this.requests.put(request);
//Only request against the childExistence mask, since the guarantee is that if childExistence bit is not set then that child is guaranteed to be empty
for (int i = 0; i < 8; i++) {
if ((childExistence&(1<<i))==0) {
//Dont watch or enqueue the child node cause it doesnt exist
continue;
}
long childPos = makeChildPos(pos, i);
request.addChildRequirement(i);
//Insert all the children into the tracking map with the node id
if (this.activeSectionMap.put(childPos, requestId|ID_TYPE_REQUEST) != NO_NODE) {
throw new IllegalStateException("Leaf request creation failed to insert child into map as a mapping already existed for the node!");
}
//Watch and request the child node at the given position
if (!this.updateRouter.watch(childPos)) {
throw new IllegalStateException("Failed to watch childPos");
}
}
this.nodeData.setNodeRequest(node, requestId);
}
private void removeSectionInternal(long position) {
int node = this.activeSectionMap.remove(position);
if (node == NO_NODE) {
@@ -177,56 +119,102 @@ public class HierarchicalNodeManager {
}
}
public void processRequestQueue(int count, long ptr) {
for (int requestIndex = 0; requestIndex < count; requestIndex++) {
int op = MemoryUtil.memGetInt(ptr + (requestIndex * 4L));
this.processRequest(op);
}
}
private void processRequest(int op) {
int node = op& NODE_ID_MSK;
if (!this.nodeData.nodeExists(node)) {
throw new IllegalStateException("Tried processing a node that doesnt exist: " + node);
}
if (this.nodeData.isNodeRequestInFlight(node)) {
throw new IllegalStateException("Tried processing a node that already has a request in flight: " + node + " pos: " + WorldEngine.pprintPos(this.nodeData.nodePosition(node)));
}
this.nodeData.markRequestInFlight(node);
//TODO: need to add a flag that says should do geometry uploads or something I.E. need to watch geometry
// and existance updates seperatly, since cull geometry
public void processResult(SectionUpdate update) {
//Need to handle cases
// geometry update, leaf node, leaf request node, internal node
//Child emptiness update!!! this is the hard bit
// if it is an internal node
// if emptiness adds node, need to then send a mesh request and wait
// when mesh result, need to remove the old child allocation block and make a new block to fit the
// new count of children
//2 branches, either its a leaf node -> emit a leaf request
// or the nodes geometry must be empty (i.e. culled from the graph/tree) so add to tracker and watch
if (this.nodeData.isLeafNode(node)) {
this.makeLeafRequest(node, this.nodeData.getNodeChildExistence(node));
} else {
//Verify that the node section is not in the section store. if it is then it is a state desynchonization
// Note that a section can be "empty" but some of its children might not be
}
}
//If the sections child existance bits fully empty, then the section should be removed
private void makeLeafRequest(int node, byte childExistence) {
long pos = this.nodeData.nodePosition(node);
//Enqueue a leaf expansion request
var request = new NodeChildRequest(pos);
int requestId = this.requests.put(request);
//Only request against the childExistence mask, since the guarantee is that if childExistence bit is not set then that child is guaranteed to be empty
for (int i = 0; i < 8; i++) {
if ((childExistence&(1<<i))==0) {
//Dont watch or enqueue the child node cause it doesnt exist
continue;
}
long childPos = makeChildPos(pos, i);
request.addChildRequirement(i);
//Insert all the children into the tracking map with the node id
if (this.activeSectionMap.put(childPos, requestId|ID_TYPE_REQUEST) != NO_NODE) {
throw new IllegalStateException("Leaf request creation failed to insert child into map as a mapping already existed for the node!");
}
//Watch and request the child node at the given position
if (!this.updateRouter.watch(childPos, WorldEngine.UPDATE_FLAGS)) {
throw new IllegalStateException("Failed to watch childPos");
}
}
this.nodeData.setNodeRequest(node, requestId);
}
final long position = update.position();
final var geometryData = update.geometry();
public void processChildChange(long position, byte childExistence) {
int nodeId = this.activeSectionMap.get(position);
if (nodeId == NO_NODE) {
System.err.println("Received update for section " + WorldEngine.pprintPos(position) + " however section position not in active in map! discarding");
//Not tracked or mapped to a node!, discard it, it was probably in progress when it was removed from the map
if (geometryData != null) {
geometryData.free();
}
System.err.println("Received child change for section " + WorldEngine.pprintPos(position) + " however section position not in active in map! discarding");
} else {
}
}
public void processGeometryResult(BuiltSection section) {
final long position = section.position;
int nodeId = this.activeSectionMap.get(position);
if (nodeId == NO_NODE) {
System.err.println("Received geometry for section " + WorldEngine.pprintPos(position) + " however section position not in active in map! discarding");
//Not tracked or mapped to a node!, discard it, it was probably in progress when it was removed from the map
section.free();
} else {
//TODO! need to not do this as it may have child data assocaited, should allocate when initally adding the TLN
if (nodeId == SENTINAL_TOP_NODE_INFLIGHT) {
//Special state for top level nodes that are in flight
if (geometryData == null) {
//FIXME: this is a bug, as the child existence could change and have an update sent, resulting in a desync
System.err.println("Top level inflight node " + WorldEngine.pprintPos(position) + " got a child msk update but was still in flight! discarding update");
return;
}
//Allocate a new node id
nodeId = this.nodeData.allocate();
this.activeSectionMap.put(position, nodeId|ID_TYPE_TOP);
int geometry = -1;
if (!geometryData.isEmpty()) {
geometry = this.geometryManager.uploadSection(geometryData);
if (!section.isEmpty()) {
geometry = this.geometryManager.uploadSection(section);
} else {
geometryData.free();
section.free();
}
this.fillNode(nodeId, position, geometry, update.childExistence());
this.fillNode(nodeId, position, geometry, (byte) 0);//INCORRECT
} else {
int type = (nodeId & ID_TYPE_MSK);
nodeId &= ~ID_TYPE_MSK;
if (type == ID_TYPE_REQUEST) {
this.requestDataUpdate(nodeId, update);
this.requestDataUpdate(nodeId);
} else if (type == ID_TYPE_NONE || type == ID_TYPE_TOP) {
//Not part of a request, just a node update,
@@ -245,7 +233,7 @@ public class HierarchicalNodeManager {
this.nodeData.setNodeChildExistence(node, childExistence);
}
private void requestDataUpdate(int nodeId, SectionUpdate update) {
private void requestDataUpdate(int nodeId) {
var request = this.requests.get(nodeId);
//Update for section part of a request, the request may be a leaf request update or an inner node update

36
src/main/java/me/cortex/voxy/common/thread/QueuedServiceSlice.java Normal file
View File

@@ -0,0 +1,36 @@
package me.cortex.voxy.common.thread;
import me.cortex.voxy.common.util.TrackedObject;
import me.cortex.voxy.common.world.WorldSection;
import net.minecraft.client.MinecraftClient;
import net.minecraft.text.Text;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BooleanSupplier;
import java.util.function.Consumer;
import java.util.function.Supplier;
public class QueuedServiceSlice <T> extends ServiceSlice {
private final ConcurrentLinkedDeque<T> queue = new ConcurrentLinkedDeque<>();
QueuedServiceSlice(ServiceThreadPool threadPool, Supplier<Consumer<T>> workerGenerator, String name, int weightPerJob, BooleanSupplier condition) {
super(threadPool, null, name, weightPerJob, condition);
//Fuck off java with the this bullshit before super constructor, fucking bullshit
super.setWorkerGenerator(() -> {
var work = workerGenerator.get();
return () -> work.accept(this.queue.pop());
});
}
@Override
public void execute() {
throw new IllegalStateException("Cannot call .execute() on a QueuedServiceSlice");
}
public void enqueue(T obj) {
this.queue.add(obj);
super.execute();
}
}

8
src/main/java/me/cortex/voxy/common/thread/ServiceSlice.java
View File

@@ -14,7 +14,7 @@ public class ServiceSlice extends TrackedObject {
final int weightPerJob;
volatile boolean alive = true;
private final ServiceThreadPool threadPool;
private final Supplier<Runnable> workerGenerator;
private Supplier<Runnable> workerGenerator;
final Semaphore jobCount = new Semaphore(0);
private final Runnable[] runningCtxs;
private final AtomicInteger activeCount = new AtomicInteger();
@@ -25,9 +25,13 @@ public class ServiceSlice extends TrackedObject {
this.threadPool = threadPool;
this.condition = condition;
this.runningCtxs = new Runnable[threadPool.getThreadCount()];
this.workerGenerator = workerGenerator;
this.name = name;
this.weightPerJob = weightPerJob;
this.setWorkerGenerator(workerGenerator);
}
protected void setWorkerGenerator(Supplier<Runnable> workerGenerator) {
this.workerGenerator = workerGenerator;
}
boolean doRun(int threadIndex) {

8
src/main/java/me/cortex/voxy/common/thread/ServiceThreadPool.java
View File

@@ -37,13 +37,17 @@ public class ServiceThreadPool {
}
public synchronized ServiceSlice createService(String name, int weight, Supplier<Runnable> workGenerator, BooleanSupplier executionCondition) {
var service = new ServiceSlice(this, workGenerator, name, weight, executionCondition);
this.insertService(service);
return service;
}
private void insertService(ServiceSlice service) {
var current = this.serviceSlices;
var newList = new ServiceSlice[current.length + 1];
System.arraycopy(current, 0, newList, 0, current.length);
var service = new ServiceSlice(this, workGenerator, name, weight, executionCondition);
newList[current.length] = service;
this.serviceSlices = newList;
return service;
}
synchronized void removeService(ServiceSlice service) {

39
src/main/java/me/cortex/voxy/common/util/MessageQueue.java Normal file
View File

@@ -0,0 +1,39 @@
package me.cortex.voxy.common.util;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.function.Consumer;
public class MessageQueue <T> {
private final Consumer<T> consumer;
private final ConcurrentLinkedDeque<T> queue = new ConcurrentLinkedDeque<>();
public MessageQueue(Consumer<T> consumer) {
this.consumer = consumer;
}
public void push(T obj) {
this.queue.add(obj);
}
public int consume() {
return this.consume(Integer.MAX_VALUE);
}
public int consume(int max) {
int i = 0;
while (i < max) {
var entry = this.queue.poll();
if (entry == null) return i;
i++;
this.consumer.accept(entry);
}
return i;
}
public final void clear(Consumer<T> cleaner) {
while (!this.queue.isEmpty()) {
cleaner.accept(this.queue.pop());
}
}
}