Difference between revisions of "GRS2"

List<PortRange> ports=new ArrayList<PortRange>(); //The ports that the TCPConnection manager should use
ports.add(new PortRange(3000, 3050));             //Any in the range between 3000 and 3050
ports.add(new PortRange(3055, 3055));             //Port 3055
TCPConnectionManager.Init(
  new TCPConnectionManagerConfig("localhost", //The hostname by which the machine is reachable 
    ports,                                    //The ports that can be used by the connection manager
    true                                      //If no port ranges were provided, or none of them could be used, use a random available port
));
TCPConnectionManager.RegisterEntry(new TCPConnectionHandler());      //Register the handler for the gRS2 incoming requests
TCPConnectionManager.RegisterEntry(new TCPStoreConnectionHandler()); //Register the handler for the gRS2 store incoming requests

//The gRS will contain only one type of records which in turn contains only a single field 
RecordDefinition[] defs=new RecordDefinition[]{          //A gRS can contain a number of different record definitions
    new GenericRecordDefinition((new FieldDefinition[] { //A record can contain a number of different field definitions
    new StringFieldDefinition("ThisIsTheField")          //The definition of the field
  }))
};

writer=new RecordWriter<GenericRecord>(
  proxy, //The proxy that defines the way the writer can be accessed
  defs   //The definitions of the records the gRS handles
);
 
writer=new RecordWriter<GenericRecord>(
  proxy, //The proxy that defines the way the writer can be accessed
  defs,  //The definitions of the records the gRS handles
  50,    //The capacity of the underlying synchronization buffer
  2,     //The maximum number of parallel records that can be concurrently accessed on partial transfer  
  0.5f   //The maximum fraction of the buffer that should be transfered during mirroring  
);
 
writer=new RecordWriter<GenericRecord>(
  proxy,           //The proxy that defines the way the writer can be accessed
  defs,            //The definitions of the records the gRS handles
  50,              //The capacity of the underlying synchronization buffer
  2,               //The maximum number of parallel records that can be concurrently accessed on partial transfer  
  0.5f,            //The maximum fraction of the buffer that should be transfered during mirroring 
  60,              //The timeout in time units after which an inactive gRS can be disposed  
  TimeUnit.SECONDS //The time unit in timeout after which an inactive gRS can be disposed
);

import java.net.URI;
import java.util.concurrent.TimeUnit;
import gr.uoa.di.madgik.grs.buffer.IBuffer.Status;
import gr.uoa.di.madgik.grs.proxy.IWriterProxy;
import gr.uoa.di.madgik.grs.record.GenericRecord;
import gr.uoa.di.madgik.grs.record.GenericRecordDefinition;
import gr.uoa.di.madgik.grs.record.RecordDefinition;
import gr.uoa.di.madgik.grs.record.field.Field;
import gr.uoa.di.madgik.grs.record.field.FieldDefinition;
import gr.uoa.di.madgik.grs.record.field.StringField;
import gr.uoa.di.madgik.grs.record.field.StringFieldDefinition;
import gr.uoa.di.madgik.grs.writer.GRS2WriterException;
import gr.uoa.di.madgik.grs.writer.RecordWriter;
 
public class StringWriter extends Thread
{
  private RecordWriter<GenericRecord> writer=null;
 
  public StringWriter(IWriterProxy proxy) throws GRS2WriterException
  {
    //The gRS will contain only one type of records which in turn contains only a single field 
    RecordDefinition[] defs=new RecordDefinition[]{          //A gRS can contain a number of different record definitions
        new GenericRecordDefinition((new FieldDefinition[] { //A record can contain a number of different field definitions
        new StringFieldDefinition("ThisIsTheField")          //The definition of the field
      }))
    };
    writer=new RecordWriter<GenericRecord>(
        proxy, //The proxy that defines the way the writer can be accessed
        defs   //The definitions of the records the gRS handles
      );
    }
 
    public URI getLocator() throws GRS2WriterException
    {
      return writer.getLocator();
    }
 
    public void run()
    {
      try
      {
        for(int i=0;i<500;i+=1)
        {
          //while the reader hasn't stopped reading
          if(writer.getStatus()!=Status.Open) break;
          GenericRecord rec=new GenericRecord();
          //Only a string field is added to the record as per definition
          rec.setFields(new Field[]{new StringField("Hello world "+i)});
          //if the buffer is in maximum capacity for the specified interval don;t wait any more
          if(!writer.put(rec,60,TimeUnit.SECONDS)) break;
        }
        //if the reader hasn't already disposed the buffer, close to notify reader that no more will be provided 
        if(writer.getStatus()!=Status.Dispose) writer.close();
      }catch(Exception ex)
      {
        ex.printStackTrace();
      }
    }
}

import gr.uoa.di.madgik.grs.reader.ForwardReader;
import gr.uoa.di.madgik.grs.reader.GRS2ReaderException;
import gr.uoa.di.madgik.grs.record.GenericRecord;
import gr.uoa.di.madgik.grs.record.field.StringField;
import java.net.URI;
 
public class StringReader extends Thread
{
  ForwardReader<GenericRecord> reader=null;
 
  public StringReader(URI locator) throws GRS2ReaderException
  {
    reader=new ForwardReader<GenericRecord>(locator);
  }
 
  public void run()
  {
    try
    {
      for(GenericRecord rec : reader)
      {
        //In case a timeout occurs while optimistically waiting for more records form an originally open writer
        if(rec==null) break;
        //Retrieve the required field of the type available in the gRS definitions
        System.out.println(((StringField)rec.getField("ThisIsTheField")).getPayload());
      }
      //Close the reader to release and dispose any resources in boith reader and writer sides
      reader.close();
    }catch(Exception ex)
    {
      ex.printStackTrace();
    }
  }
}

import java.util.ArrayList;
import java.util.List;
import gr.uoa.di.madgik.commons.server.PortRange;
import gr.uoa.di.madgik.commons.server.TCPConnectionManager;
import gr.uoa.di.madgik.commons.server.TCPConnectionManagerConfig;
import gr.uoa.di.madgik.grs.proxy.tcp.TCPConnectionHandler;
import gr.uoa.di.madgik.grs.proxy.tcp.TCPStoreConnectionHandler;
import gr.uoa.di.madgik.grs.proxy.tcp.TCPWriterProxy;
 
public class MainTest
{
  public static void main(String []args) throws Exception
  {
    List<PortRange> ports=new ArrayList<PortRange>(); //The ports that the TCPConnection manager should use
    ports.add(new PortRange(3000, 3050));             //Any in the range between 3000 and 3050
    ports.add(new PortRange(3055, 3055));             //Port 3055
    TCPConnectionManager.Init(
      new TCPConnectionManagerConfig("localhost", //The hostname by which the machine is reachable 
        ports,                                    //The ports that can be used by the connection manager
        true                                      //If no port ranges were provided, or none of them could be used, use a random available port
    ));
    TCPConnectionManager.RegisterEntry(new TCPConnectionHandler());      //Register the handler for the gRS2 incoming requests
    TCPConnectionManager.RegisterEntry(new TCPStoreConnectionHandler()); //Register the handler for the gRS2 store incoming requests
 
    StringWriter writer=new StringWriter(new TCPWriterProxy());
    StringReader reader=new StringReader(writer.getLocator());
 
    writer.start();
    reader.start();
 
    writer.join();
    reader.join();
  }
}