package com.github.shredder121.asyncaudio.common;

import java.net.DatagramPacket;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import lombok.experimental.NonFinal;

public class ProvideForkJoinTask extends ForkJoinTask<DatagramPacket> {

	Supplier<DatagramPacket> provider;
	BlockingQueue<DatagramPacket> queue;

	/**
	 * A reference to a {@link DatagramPacket} that couldn't be offered.
	 *
	 * <p>
	 *   Set to a non-null value when a round needs to be skipped.
	 *   The packet will then be picked up on a subsequent run.
	 * </p>
	 */
	AtomicReference<DatagramPacket> packetRef = new AtomicReference<>();


	@NonFinal //flag to stop
	private volatile boolean stopRequested; //false

	public ProvideForkJoinTask(Supplier<DatagramPacket> provider, BlockingQueue<DatagramPacket> queue) {
		this.provider = provider;
		this.queue = queue;
	}

    @Override
    protected boolean exec() {
		DatagramPacket packet = this.packetRef.getAndSet(null);
		if (this.stopRequested) {
			return true;
		} else if (packet == null) {
			packet = this.provider.get();
		}

        if (packet == null || !this.queue.offer(packet)) {
		    // offer failed, retry in next run
			this.packetRef.set(packet);
			ForkJoinPool pool = getPool();
			ForkJoinTask<?> me = this;

			// instead of spinning, suspend execution
			reinitialize();
			CommonAsync.rescheduler.schedule(() -> pool.execute(me), 40, TimeUnit.MILLISECONDS);
			return false;
		}

        reinitialize();
		fork();
        return false;
    }

    @Override
    protected void setRawResult(DatagramPacket value) {
		throw new UnsupportedOperationException("Not needed.");
    }

	@Override
	public DatagramPacket getRawResult() {
		throw new UnsupportedOperationException("Not needed.");
	}

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        this.stopRequested = true;
        return super.cancel(mayInterruptIfRunning);
    }
}


@@ -813,17 +813,16 @@
 e%3C%3E();%0A%0A
-%0A
 %09@NonFin
@@ -1041,28 +1041,25 @@
  queue;%0A%09%7D%0A%0A
-    
+%09
 @Override%0A  
@@ -1048,36 +1048,33 @@
 %0A%09%7D%0A%0A%09@Override%0A
-    
+%09
 protected boolea
@@ -1225,24 +1225,43 @@
 %09%09%09packet = 
+optionallyCopyData(
 this.provide
@@ -1267,31 +1267,26 @@
 er.get()
+)
 ;%0A%09%09%7D%0A%0A
-        
+%09%09
 if (pack
@@ -1330,20 +1330,17 @@
 t)) %7B%0A%09%09
-    
+%09
 // offer
@@ -1358,11 +1358,8 @@
 try 
-in 
 next
@@ -1633,24 +1633,18 @@
 e;%0A%09%09%7D%0A%0A
-        
+%09%09
 reinitia
@@ -1665,41 +1665,661 @@
 ();%0A
-        return false;%0A    %7D%0A%0A    
+%09%09return false;%0A%09%7D%0A%0A%09/**%0A%09 * Makes a copy of the data in the packet to make the packet safe to store for longer periods of time.%0A%09 *%0A%09 * %3Cp%3E%0A%09 *   JDA changed to have only one backing array for all packets, to reduce allocations.%0A%09 *   This means that in JAPP code we have to copy the packet data before storing it for longer periods of time.%0A%09 * %3C/p%3E%0A%09 *%0A%09 * @param packet the packet to copy the data for%0A%09 * @return the adjusted packet%0A%09 */%0A%09private static DatagramPacket optionallyCopyData(DatagramPacket packet) %7B%0A%09%09if (packet != null) %7B%0A%09%09%09packet.setData(packet.getData().clone(), packet.getOffset(), packet.getLength());%0A%09%09%7D%0A%09%09return packet;%0A%09%7D%0A%0A%09
 @Ove
@@ -2324,20 +2324,17 @@
 verride%0A
-    
+%09
 protecte
@@ -2435,20 +2435,17 @@
 ded.%22);%0A
-    
+%09
 %7D%0A%0A%09@Ove
@@ -2552,20 +2552,17 @@
 %22);%0A%09%7D%0A%0A
-    
+%09
 @Overrid
@@ -2563,20 +2563,17 @@
 verride%0A
-    
+%09
 public b
@@ -2619,24 +2619,18 @@
 ning) %7B%0A
-        
+%09%09
 this.sto
@@ -2652,16 +2652,10 @@
 ue;%0A
-        
+%09%09
 retu
@@ -2698,13 +2698,9 @@
 g);%0A
-    
+%09
 %7D%0A%7D%0A
-%0A

Copy the data in the packet before storing it in the queue