Skip to content

feat(appeng): add fluid support and network_changed event subscription and fix thread-safety issues in allItems. #169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hinyb wants to merge 5 commits into
base: master
Choose a base branch
from
Open

feat(appeng): add fluid support and network_changed event subscription and fix thread-safety issues in allItems. #169

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ee06bb2
feat(appeng): add allFluids and network_changed event subscription.
hinyb Dec 28, 2025
cde2eb3
fix: resolve thread-safety issues in allItems and allFluids.
hinyb Dec 31, 2025
8e830a5
feat(appeng): add fluid support.
hinyb Jan 6, 2026
01e3771
Merge branch 'master' into appeng
hinyb Jan 26, 2026
7b6ba23
fix: fluid stacks and NetworkContents failing to persist.
hinyb Jan 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
219 changes: 155 additions & 64 deletions src/main/scala/li/cil/oc/integration/appeng/NetworkControl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,25 +1,23 @@
package li.cil.oc.integration.appeng

import java.lang
import appeng.api.AEApi
import appeng.api.config.Actionable
import appeng.api.networking.IGridNode
import appeng.api.networking.crafting.{CraftingItemList, ICraftingLink, ICraftingRequester}
import appeng.api.networking.security.{BaseActionSource, IActionHost, MachineSource}
import appeng.api.networking.storage.IBaseMonitor
import appeng.api.storage.IMEMonitorHandlerReceiver
import appeng.api.storage.data.{IAEFluidStack, IAEItemStack, IItemList}
import appeng.api.storage.data.{IAEFluidStack, IAEItemStack, IAEStack, IItemList}
import appeng.api.storage.{IMEMonitor, IMEMonitorHandlerReceiver}
import appeng.api.util.AECableType
import appeng.me.GridAccessException
import appeng.me.cluster.implementations.CraftingCPUCluster
import appeng.me.helpers.IGridProxyable
import appeng.tile.crafting.TileCraftingMonitorTile
import appeng.util.IterationCounter
import appeng.util.item.AEItemStack
import appeng.util.{IterationCounter, Platform}
import appeng.util.item.{AEFluidStack, AEItemStack}
import com.google.common.collect.ImmutableSet
import li.cil.oc.OpenComputers
import li.cil.oc.api.machine.Arguments
import li.cil.oc.api.machine.Callback
import li.cil.oc.api.machine.Context
import li.cil.oc.api.machine.{Arguments, Callback, Context}
import li.cil.oc.api.network.Node
import li.cil.oc.api.prefab.AbstractValue
import li.cil.oc.common.EventHandler
Expand All @@ -38,11 +36,14 @@ import net.minecraft.tileentity.TileEntity
import net.minecraftforge.common.DimensionManager
import net.minecraftforge.common.util.Constants.NBT
import net.minecraftforge.common.util.ForgeDirection
import net.minecraftforge.fluids.{FluidRegistry, FluidStack}

import java.lang
import javax.annotation.Nonnull
import scala.collection.convert.WrapAsJava._
import scala.collection.convert.WrapAsScala._
import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.language.existentials
Expand All @@ -54,9 +55,13 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi

def node: Node

private def allItems: Iterable[IAEItemStack] = tile.getProxy.getStorage.getItemInventory.getStorageList
private def getAvailableItem(@Nonnull request: IAEItemStack, iteration: Int): IAEItemStack = tile.getProxy.getStorage.getItemInventory.getAvailableItem(request, iteration)
private def allCraftables: Iterable[IAEItemStack] = allItems.collect{ case aeItem if aeItem.isCraftable => aeCraftItem(aeItem, tile) }
private def allItems: IItemList[IAEItemStack] = tile.getProxy.getStorage.getItemInventory.getStorageList

private def getAvailableItem(@Nonnull request: IAEItemStack, iteration: Int = IterationCounter.fetchNewId()): IAEItemStack = tile.getProxy.getStorage.getItemInventory.getAvailableItem(request, iteration)

private def allFluids: IItemList[IAEFluidStack] = tile.getProxy.getStorage.getFluidInventory.getStorageList

private def getAvailableFluid(@Nonnull request: IAEFluidStack, iteration: Int = IterationCounter.fetchNewId()): IAEFluidStack = tile.getProxy.getStorage.getFluidInventory.getAvailableItem(request, iteration)

@Callback(doc = "function():table -- Get a list of tables representing the available CPUs in the network.")
def getCpus(context: Context, args: Arguments): Array[AnyRef] = {
Expand All @@ -80,9 +85,18 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi
val filter = args.optTable(0, Map.empty[AnyRef, AnyRef]).collect {
case (key: String, value: AnyRef) => (key, value)
}
result(allCraftables
.collect{ case aeCraftItem if filter.isEmpty || matches(convert(aeCraftItem, tile), filter) => new NetworkControl.Craftable(tile, aeCraftItem) }
.toArray)

val builder = mutable.ArrayBuilder.make[AnyRef]
(allItems.view ++ allFluids.view).foreach({ s =>
if (s.isCraftable) {
val c = asCraft(s, tile).asInstanceOf[IAEStack[_ <: IAEStack[_]]]
if (matches(convert(c, tile), filter)) {
builder += new Craftable(tile, c)
}
}
})

result(builder.result())
}

@Callback(doc = "function([filter:table]):table -- Get a list of the stored items in the network.")
Expand Down Expand Up @@ -133,9 +147,26 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi
.toArray)
}

@Callback(doc = "function():userdata -- Get an iterator object for the list of the items in the network.")
@Callback(doc = "function([name:string]):table -- Get a list of the stored fluids in the network.")
def getFluidInNetwork(context: Context, args: Arguments): Array[AnyRef] = {
FluidRegistry.getFluid(args.checkString(0)) match {
case null => result(null)
case fluid =>
getAvailableFluid(AEFluidStack.create(new FluidStack(fluid, 0))) match {
case null => result(null)
case fluid => result(convert(fluid, tile))
}
}
}

@Callback(doc = "function([subscribe:bool]):userdata -- Get an iterator object for the list of the items in the network. [Event: network_item_changed]")
def allItems(context: Context, args: Arguments): Array[AnyRef] = {
result(new NetworkContents(tile))
result(new ItemNetworkContents(tile, node, args.optBoolean(0, false)))
}

@Callback(doc = "function([subscribe:bool]):userdata -- Get an iterator object for the list of the fluids in the network. [Event: network_fluid_changed]")
def allFluids(context: Context, args: Arguments): Array[AnyRef] = {
result(new FluidNetworkContents(tile, node, args.optBoolean(0, false)))
}

@Callback(doc = "function(filter:table, dbAddress:string[, startSlot:number[, count:number]]): Boolean -- Store items in the network matching the specified filter in the database with the specified address.")
Expand All @@ -145,7 +176,7 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi
}
DatabaseAccess.withDatabase(node, args.checkString(1), database => {
val items = allItems
.collect{ case aeItem if matches(convert(aeItem, tile), filter) => aePotentialItem(aeItem, tile)}.toArray
.collect { case aeItem if matches(convert(aeItem, tile), filter) => aePotentialItem(aeItem, tile) }.toArray
val offset = args.optSlot(database.data, 2, 0)
val count = args.optInteger(3, Int.MaxValue) min (database.size - offset) min items.length
var slot = offset
Expand All @@ -167,7 +198,7 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi

@Callback(doc = "function():table -- Get a list of the stored fluids in the network.")
def getFluidsInNetwork(context: Context, args: Arguments): Array[AnyRef] =
result(tile.getProxy.getStorage.getFluidInventory.getStorageList.filter(isFluidVisible).map(fs => convert(fs, tile)).toArray)
result(allFluids.filter(isFluidVisible).map(fs => convert(fs, tile)).toArray)

@Callback(doc = "function():number -- Get the average power injection into the network.")
def getAvgPowerInjection(context: Context, args: Arguments): Array[AnyRef] =
Expand All @@ -192,7 +223,7 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi
private def matches(stack: java.util.HashMap[String, AnyRef], filter: scala.collection.mutable.Map[String, AnyRef]): Boolean = {
if (stack == null) return false
filter.forall {
case (key: String, value: AnyRef) =>
case (key: String, value: AnyRef) =>
val stack_value = stack.get(key)
if (stack_value == null) false
else value match {
Expand All @@ -209,7 +240,7 @@ trait NetworkControl[AETile >: Null <: TileEntity with IGridProxyable with IActi
object NetworkControl {

//noinspection ScalaUnusedSymbol
class Craftable(var controller: TileEntity with IGridProxyable with IActionHost, var stack: IAEItemStack) extends AbstractValue with ICraftingRequester {
class Craftable(var controller: TileEntity with IGridProxyable with IActionHost, var stack: IAEStack[_ <: IAEStack[_]]) extends AbstractValue with ICraftingRequester {
def this() = this(null, null)

private val links = mutable.Set.empty[ICraftingLink]
Expand All @@ -236,7 +267,10 @@ object NetworkControl {
// ----------------------------------------------------------------------- //

@Callback(doc = "function():table -- Returns the item stack representation of the crafting result.")
def getItemStack(context: Context, args: Arguments): Array[AnyRef] = Array(stack.getItemStack)
def getStack(context: Context, args: Arguments): Array[AnyRef] = stack match {
case item: IAEItemStack => result(item.getItemStack)
case fluid: IAEFluidStack => result(fluid.getFluidStack)
}

@Callback(doc = "function([amount:int[, prioritizePower:boolean[, cpuName:string]]]):userdata -- Requests the item to be crafted, returning an object that allows tracking the crafting status.")
def request(context: Context, args: Arguments): Array[AnyRef] = {
Expand All @@ -245,7 +279,7 @@ object NetworkControl {
}

val count = args.optInteger(0, 1)
val request = stack.copy
val request = stack.copy.asInstanceOf[IAEStack[_]]
request.setStackSize(count)

val craftingGrid = controller.getProxy.getCrafting
Expand Down Expand Up @@ -297,22 +331,27 @@ object NetworkControl {

override def load(nbt: NBTTagCompound) {
super.load(nbt)
stack = AEItemStack.loadItemStackFromNBT(nbt)
if (nbt.hasKey("StackType")) {
stack = Platform.readStackNBT(nbt, true).asInstanceOf[IAEStack[_ <: IAEStack[_]]]
}
else {
stack = AEItemStack.loadItemStackFromNBT(nbt)
}
loadController(nbt, c => controller = c)
links ++= nbt.getTagList("links", NBT.TAG_COMPOUND).map(
(nbt: NBTTagCompound) => AEApi.instance.storage.loadCraftingLink(nbt, this))
}

override def save(nbt: NBTTagCompound) {
super.save(nbt)
stack.writeToNBT(nbt)
Platform.writeStackNBT(stack, nbt, true)
saveController(controller, nbt)
nbt.setNewTagList("links", links.map(_.writeToNBT _))
}
}

//noinspection ScalaUnusedSymbol
class Cpu(var controller: TileEntity with IGridProxyable, var index : Int, var cpu : CraftingCPUCluster) extends AbstractValue {
class Cpu(var controller: TileEntity with IGridProxyable, var index: Int, var cpu: CraftingCPUCluster) extends AbstractValue {
def this() = this(null, 0, null)

@Callback(doc = "function():boolean -- Cancel this CPU current crafting job.")
Expand Down Expand Up @@ -369,7 +408,7 @@ object NetworkControl {
result(aeStack.getItemStack)
}
}

private def getCpu = {
if (cpu == null && controller != null) {
var i = 0
Expand All @@ -390,6 +429,7 @@ object NetworkControl {
nbt.setInteger("index", index)
saveController(controller, nbt)
}

override def load(nbt: NBTTagCompound) {
super.load(nbt)
index = nbt.getInteger("index")
Expand Down Expand Up @@ -453,44 +493,50 @@ object NetworkControl {
}

//noinspection ConvertNullInitializerToUnderscore
class NetworkContents(var controller: TileEntity with IGridProxyable with IActionHost) extends AbstractValue with IMEMonitorHandlerReceiver[IAEItemStack]
{
def this() = this(null)
if (controller != null)
controller.getProxy.getStorage.getItemInventory.addListener(this, null)
private var addedListener = true
private var items : IItemList[IAEItemStack] = null
private var itemIterator : java.util.Iterator[IAEItemStack] = null
abstract class NetworkContents[T <: IAEStack[T]](var controller: TileEntity with IGridProxyable with IActionHost, var node: Node, var subscribe: Boolean) extends AbstractValue with IMEMonitorHandlerReceiver[T] {
def this() = this(null, null, false)

def getInventory: IMEMonitor[T]

def withInventory[R](action: IMEMonitor[T] => R): Option[R] = {
for {
_ <- Option(controller)
inv <- Option(getInventory)
} yield action(inv)
}

def convertItem(stack: IAEStack[_]): java.util.HashMap[String, AnyRef]

val event_name: String = "network_changed"

private var items: IItemList[T] = null
private var itemIterator: java.util.Iterator[T] = null
private var index = 0

private def updateItems(): Unit = {
val list: Option[IItemList[T]] = withInventory(_.getStorageList)
items = list.orNull
if (items != null) {
itemIterator = items.iterator
itemIterator.drop(index)
}
}

updateItems()
withInventory(_.addListener(this, null))

override def call(context: Context, arguments: Arguments): Array[AnyRef] = {
if (controller == null)
if (controller == null || itemIterator == null)
return null
if (!addedListener) {
controller.getProxy.getStorage.getItemInventory.addListener(this, null)
addedListener = true
}
if (items == null) {
items = controller.getProxy.getStorage.getItemInventory.getStorageList
if (items != null)
itemIterator = items.iterator
if (itemIterator != null)
for (_ <- 1 to index) {
if (itemIterator.hasNext)
itemIterator.next
}
}
if (this.itemIterator == null && this.items != null)
this.itemIterator = items.iterator
if (!this.itemIterator.hasNext)
return null
index += 1
Array[AnyRef](convert(itemIterator.next(), controller))
result(convertItem(itemIterator.next()))
}

override def load(nbt: NBTTagCompound) {
super.load(nbt)
addedListener = false
withInventory(_.addListener(this, null))
loadController(nbt, c => controller = c)
}

Expand All @@ -509,26 +555,64 @@ object NetworkControl {
override def isValid(verificationToken: Any): Boolean = valid

override def onListUpdate(): Unit = {
this.items = null
updateItems()
}

override def postChange(monitor: IBaseMonitor[T], change: lang.Iterable[T], actionSource: BaseActionSource): Unit = {
if (subscribe && node != null) {
val flatArgs = ArrayBuffer[Object](event_name)
flatArgs ++= change.map(convertItem(_))
node.sendToReachable("computer.signal", flatArgs: _*)
}
updateItems()
}
}

class ItemNetworkContents(controller: TileEntity with IGridProxyable with IActionHost, node: Node, subscribe: Boolean) extends NetworkContents[IAEItemStack](controller, node, subscribe) {
def this() = this(null, null, false)

override val event_name: String = "network_item_changed"

def getInventory: IMEMonitor[IAEItemStack] = {
try return controller.getProxy.getStorage.getItemInventory
catch {
case ignored: GridAccessException =>
}
return null
}
override def postChange(monitor: IBaseMonitor[IAEItemStack], change: lang.Iterable[IAEItemStack], actionSource: BaseActionSource): Unit = {
this.items = null

override def convertItem(stack: IAEStack[_]): java.util.HashMap[String, AnyRef] = convert(stack.asInstanceOf[IAEItemStack], controller)
}

class FluidNetworkContents(controller: TileEntity with IGridProxyable with IActionHost, node: Node, subscribe: Boolean) extends NetworkContents[IAEFluidStack](controller, node, subscribe) {
def this() = this(null, null, false)

override val event_name: String = "network_fluid_changed"

override def getInventory: IMEMonitor[IAEFluidStack] = {
try return controller.getProxy.getStorage.getFluidInventory
catch {
case ignored: GridAccessException =>
}
return null
}

override def convertItem(stack: IAEStack[_]): java.util.HashMap[String, AnyRef] = convert(stack.asInstanceOf[IAEFluidStack], controller)
}

def aeCraftItem(aeItem: IAEItemStack, tile: TileEntity with IGridProxyable): IAEItemStack = {
val patterns = tile.getProxy.getCrafting.getCraftingFor(aeItem, null, 0, tile.getWorldObj)
patterns.find(pattern => pattern.getOutputs.exists(_.isSameType(aeItem))) match {
case Some(pattern) => pattern.getOutputs.find(_.isSameType(aeItem)).get
case _ => aeItem.copy.setStackSize(0) // Should not be possible, but hey...
def asCraft(stack: IAEStack[_], tile: TileEntity with IGridProxyable): IAEStack[_] = {
tile.getProxy.getCrafting.getCraftingFor(stack, null, 0, tile.getWorldObj).view.map(a => a.getAEOutputs.find(_.isSameType(stack)).get).headOption.getOrElse[IAEStack[_]] {
val result = stack.copy().asInstanceOf[IAEStack[_ <: IAEStack[_]]]
result.setStackSize(0)
result
}
}

def aePotentialItem(aeItem: IAEItemStack, tile: TileEntity with IGridProxyable): IAEItemStack = {
if (aeItem.getStackSize > 0 || !aeItem.isCraftable)
aeItem
else
aeCraftItem(aeItem, tile)
asCraft(aeItem, tile).asInstanceOf[IAEItemStack]
}

def hashConvert(value: java.util.HashMap[_, _]) = {
Expand All @@ -537,10 +621,17 @@ object NetworkControl {
hash
}

def convert(aeItem: IAEStack[_ <: IAEStack[_]], tile: TileEntity with IGridProxyable): java.util.HashMap[String, AnyRef] = {
aeItem match {
case item: IAEItemStack => convert(item, tile)
case fluid: IAEFluidStack => convert(fluid, tile)
}
}

def convert(aeItem: IAEItemStack, tile: TileEntity with IGridProxyable): java.util.HashMap[String, AnyRef] = {
val potentialItem = aePotentialItem(aeItem, tile)
val result = Registry.convert(Array[AnyRef](potentialItem.getItemStack))
.collect { case hash: java.util.HashMap[_,_] => hashConvert(hash) }
.collect { case hash: java.util.HashMap[_, _] => hashConvert(hash) }
if (result.length > 0) {
val hash = result(0)
// it would have been nice to put these fields in a registry convert
Expand All @@ -564,7 +655,7 @@ object NetworkControl {
null
}

private def loadController(nbt: NBTTagCompound, f : TileEntity with IGridProxyable with IActionHost => Unit ) : Unit = {
private def loadController(nbt: NBTTagCompound, f: TileEntity with IGridProxyable with IActionHost => Unit): Unit = {
if (nbt.hasKey("dimension")) {
val dimension = nbt.getInteger("dimension")
val x = nbt.getInteger("x")
Expand Down