// Merger.java
// Michele Loreti and the jSpace Developers, 2017
// Modified by Andrew Davison, ad@coe.psu.ac.th, May 2023

/* Only one Merging thread can read from the space at once, by
   grabbing the ("MERGE") tuple. It reads in any two 
   MERGING tuples, merges the two arrays together in ascending order
   and puts out a new, bigger MERGING tuple with the new array.

   But if the new array is the size of the original unsorted 
   array, then merging is complete and the thread puts out a
   (FIN, arr) instead, which is read in by main() and reported.

  Usage:
    > compile.bat Merger.java

    Part of the application made up of:
       * SpaceSorter.java
       * Splitter.java
       * Merger.java
*/


import java.util.*;

import org.jspace.*;


public class Merger implements Runnable
{
  private Space space;
  private PrintAreas pas;
  private String title;
  private int arrSize;   // size of the complete array


  public Merger(int id, int size, PrintAreas pas, Space s)
  {
    arrSize = size;
    this.pas = pas;
    space = s;
    title = "Merger-"+(id-2); // name used in PrintAreas 
    pas.setTitle(id, title);  
  }

  public void run()
  {
    try {
      while (true) {
        /* successfully grabbing the MERGE tuple means 
           that this thread can read arrays from the space
        */
        space.get(new ActualField("MERGE"));

        // get any subarray
        Object[] parts1 = space.get( 
                     new ActualField("MERGING"),
                     new FormalField(Object.class)); // array
        int[] arr1 = (int[])parts1[1];
        pas.println(title, "Got " + Arrays.toString(arr1));

        if (arr1.length == arrSize) { // merging is complete
          space.put("FIN", arr1);
          return;    // no need to release "MERGE"
        }
        else {
          // get a second subarray
          Object[] parts2 = space.get(
                       new ActualField("MERGING"),
                       new FormalField(Object.class)); // array
          int[] arr2 = (int[])parts2[1];
          pas.println(title, "Got " + Arrays.toString(arr2));
          space.put("MERGE");  
             /* this task has finished all of its reading, 
                so release the MERGE lock for some other thread
             */
          int[] mergedArr = merge(arr1, arr2);
          // put out the new array for merging
          space.put("MERGING", mergedArr);
          pas.println(title, "Put " + Arrays.toString(mergedArr));
        }
      }
    }
    catch (Exception e) 
    { System.out.println(e);  }
  } /* run */



  private static int[] merge(int[] arr1, int[] arr2)
  // merge the two ordered arrays into ascending order
  {
    int[] mergedArr = new int[arr1.length + arr2.length];
    int i = 0, j = 0, k = 0;

    while (true) {
      if (i == arr1.length) {
        while(j < arr2.length) {
          mergedArr[k] = arr2[j];
          k++; j++;
        }
        break;
      }

      if (j == arr2.length) {
        while(i < arr1.length) {
          mergedArr[k] = arr1[i];
          k++; i++;
        }
        break;
      }

      if (arr1[i] <= arr2[j]) {
        mergedArr[k] = arr1[i];
        i++;
      }
      else {
        mergedArr[k] = arr2[j];
        j++;
      }
      k++;
    }
    return mergedArr;
  } /* merge */


}  // end of Merger class