<commit_before>/*Copyright (C) 2010-2012 Institute for Software Integrated Systems (ISIS)
This software was developed by the Institute for Software Integrated
Systems (ISIS) at Vanderbilt University, Tennessee, USA for the 
Transformative Apps program under DARPA, Contract # HR011-10-C-0175.
The United States Government has unlimited rights to this software. 
The US government has the right to use, modify, reproduce, release, 
perform, display, or disclose computer software or computer software 
documentation in whole or in part, in any manner and for any 
purpose whatsoever, and to have or authorize others to do so.
 */
package edu.vu.isis.ammo.core.distributor;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.json.JSONTokener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import android.content.ComponentName;
import android.content.ContentResolver;
import android.content.ContentUris;
import android.content.ContentValues;
import android.content.Context;
import android.content.Intent;
import android.content.ServiceConnection;
import android.content.res.AssetFileDescriptor;
import android.database.Cursor;
import android.database.sqlite.SQLiteException;
import android.net.Uri;
import android.os.IBinder;
import android.os.ParcelFileDescriptor;
import android.os.RemoteException;
import edu.vu.isis.ammo.api.IDistributorAdaptor;
import edu.vu.isis.ammo.api.type.Payload;
import edu.vu.isis.ammo.api.type.Provider;
import edu.vu.isis.ammo.core.AmmoService;
import edu.vu.isis.ammo.core.PLogger;
import edu.vu.isis.ammo.core.distributor.DistributorDataStore.DisposalState;
import edu.vu.isis.ammo.core.distributor.DistributorPolicy.Encoding;
import edu.vu.isis.ammo.core.network.AmmoGatewayMessage;

/**
 * The purpose of these objects is lazily serialize an object.
 * Once it has been serialized once a copy is kept.
 *
 */
public class RequestSerializer {
  private static final Logger logger = LoggerFactory.getLogger("dist.serializer");

  /**
   * This enumeration's codes must match those of 
   * the AmmoGen files.
   */
  public enum FieldType {
    NULL(0), 
    BOOL(1), 
    BLOB(2), 
    FLOAT(3),
    INTEGER(4), 
    LONG(5), 
    TEXT(6), 
    REAL(7),
    FK(8), 
    GUID(9), 
    EXCLUSIVE(10), 
    INCLUSIVE(11),
    TIMESTAMP(12),
    SHORT(13),
    FILE(14);

    private final int code;

    private FieldType(int code) {
      this.code = code;
    }

    private static final Map<Integer,FieldType> codemap = 
        new HashMap<Integer,FieldType>();
    static {
      for (FieldType t : FieldType.values()) {
        FieldType.codemap.put(t.code, t);
      }
    }

    public int toCode() {
      return this.code;
    }

    public static FieldType fromCode(final int code) {
      return FieldType.codemap.get(code);
    }
  }

  /**
   * The presence of the BLOB_MARKER_FIELD as the first byte in the 
   * footer for a blob data section indicates where the blob should 
   * be placed in the content provider.
   */
  public static final byte BLOB_MARKER_FIELD = (byte)0xff;

  private enum BlobTypeEnum { 
        LARGE, SMALL; 

        /**
         * The difficulty here is that the blob may have trailing null bytes.
         * e.g.
         *  fieldName = [data], fieldNameBlob= [[100, 97, 116, 97]], blob = [[100, 97, 116, 97, 0]]
         * These should result it a match.
         */
        public static BlobTypeEnum infer(String fieldName, byte[] blob) {
       
           final byte[] fieldNameBlob; 
          try {
              fieldNameBlob = fieldName.getBytes("UTF-8");
          } catch (java.io.UnsupportedEncodingException ex) {
              return SMALL;
          }

         logger.trace("processing blob fieldName = [{}], fieldNameBlob= [{}], blob = [{}]", 
             new Object[] {fieldName, fieldNameBlob, blob});

          if (blob == null) return LARGE;
          if (blob.length < 1) return LARGE;

          if (fieldNameBlob.length == blob.length)
              return Arrays.equals(blob, fieldNameBlob) ? LARGE : SMALL;

          if (fieldNameBlob.length > blob.length) return SMALL;

          int i;
          for (i = 0; i < fieldNameBlob.length; i++) {
             if (fieldNameBlob[i] != blob[i]) return SMALL;
          }
          for (; i < blob.length; i++) {
               if (blob[i] != (byte)0) return SMALL;
          }
          return LARGE;
    }
  }
  
  public interface OnReady  {
    public AmmoGatewayMessage run(Encoding encode, byte[] serialized);
  }
  public interface OnSerialize  {
    public byte[] run(Encoding encode);
  }

  public final Provider provider;
  public final Payload payload;
  private OnReady readyActor;
  private OnSerialize serializeActor;
  private AmmoGatewayMessage agm;

  /**
   * This maintains a set of persistent connections to 
   * content provider adapter services.
   */
  final static private Map<String,IDistributorAdaptor> remoteServiceMap;
  static {
    remoteServiceMap = new HashMap<String,IDistributorAdaptor>(10);
  }

  private RequestSerializer(Provider provider, Payload payload) {
    this.provider = provider;
    this.payload = payload;
    this.agm = null;


    this.readyActor = new RequestSerializer.OnReady() {
      @Override
      public AmmoGatewayMessage run(Encoding encode, byte[] serialized) {
        logger.trace("ready actor not defined {}", encode);
        return null;
      }
    };  
    this.serializeActor = new RequestSerializer.OnSerialize() {
      @Override
      public byte[] run(Encoding encode) {
        logger.trace("serialize actor not defined {}", encode);
        return null;
      }
    };
  }

  public static RequestSerializer newInstance() {
    return new RequestSerializer(null, null);
  }
  public static RequestSerializer newInstance(Provider provider, Payload payload) {
    return new RequestSerializer(provider, payload);
  }

  public DisposalState act(final AmmoService that, final Encoding encode, final String channel) {
    final RequestSerializer parent = RequestSerializer.this;
    final Encoding local_encode = encode;
    final String local_channel = channel;
    if (parent.agm == null) {
      final byte[] agmBytes = parent.serializeActor.run(local_encode);
      parent.agm = parent.readyActor.run(local_encode, agmBytes);
    }
    if (parent.agm == null)
      return null;

    return that.sendRequest(parent.agm, local_channel);
  }

  public void setAction(OnReady action) {
    this.readyActor = action;
  }

  public void setSerializer(OnSerialize onSerialize) {
    this.serializeActor = onSerialize;
  }


  public static byte[] serializeFromContentValues(ContentValues cv, final DistributorPolicy.Encoding encoding) {

    logger.trace("serializing using content values and encoding {}", encoding);
    switch (encoding.getType()) {
    case JSON: 
    {
      return encodeAsJson (cv);
    }

    case TERSE: 
    {
      // Need to be implemented ...
    }
    // TODO custom still needs a lot of work
    // It will presume the presence of a SyncAdaptor for the content provider.
    case CUSTOM:
    default:
    {
    }
    }
    return null;
  }

  private static byte[] encodeAsJson (ContentValues cv) {
    // encoding in json for now ...
    Set<Map.Entry<String, Object>> data = cv.valueSet();
    Iterator<Map.Entry<String, Object>> iter = data.iterator();       
    final JSONObject json = new JSONObject();

    while (iter.hasNext())
    {
      Map.Entry<String, Object> entry = (Map.Entry<String, Object>)iter.next();         
      try {
        if (entry.getValue() instanceof String)
          json.put(entry.getKey(), cv.getAsString(entry.getKey()));
        else if (entry.getValue() instanceof Integer)
          json.put(entry.getKey(), cv.getAsInteger(entry.getKey()));
      } catch (JSONException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
      }
    }

    return json.toString().getBytes();
  }

  /**
   * This simple Future which can only be bound to a value once.
   */
  public static class DataFlow<V> implements Future<V> {
    final AtomicReference<V> value;

    public DataFlow() {
      this.value = new  AtomicReference<V>(); 
    }

    public DataFlow(V value) {
      this();
      this.value.set(value);
    }

    public void bind(final V value) {
      if (! this.value.compareAndSet(null, value)) return;
      this.notifyAll();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
      return false;
    }

    @Override
    public V get() throws InterruptedException, ExecutionException {
      synchronized (this) {
        if (this.value.get() == null) {
          this.wait();
        }
      }
      return this.value.get();
    }

    @Override
    public V get(long timeout, TimeUnit unit) 
        throws InterruptedException, ExecutionException, TimeoutException {
      synchronized (this) {
        if (this.value.get() == null) {
          this.wait(unit.toMillis(timeout));
        }
      }
      return this.value.get();
    }

    @Override
    public boolean isCancelled() {
      return false;
    }

    @Override
    public boolean isDone() {
      return this.value.get() != null;
    }
  }


  /**
   * All serializer methods return this dataflow variable.
   */
  public static class ByteBufferFuture extends DataFlow<ByteBuffer> {

    public ByteBufferFuture() {
      super();
    }

    public ByteBufferFuture(ByteBuffer value) {
      super(value);
    }

    public static ByteBufferFuture wrap(byte[] value) {
      return new ByteBufferFuture(ByteBuffer.wrap(value));
    }
  }
  /**
   *
   */  

  public static byte[] serializeFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    logger.trace("serializing using encoding {}", encoding);
    final ByteBufferFuture result;
    final Encoding.Type encodingType = encoding.getType();
    switch (encodingType) {
    case CUSTOM:
      result = RequestSerializer.serializeCustomFromProvider(resolver, tupleUri, encoding);
      break;
    case JSON: 
      result = RequestSerializer.serializeJsonFromProvider(resolver, tupleUri, encoding);
      break;
    case TERSE: 
      result = RequestSerializer.serializeTerseFromProvider(resolver, tupleUri, encoding);
      break;
    default:
      result = RequestSerializer.serializeCustomFromProvider(resolver, tupleUri, encoding);
    }
    try {
      return result.get().array();
    } catch (InterruptedException ex) {
      ex.printStackTrace();
    } catch (ExecutionException ex) {
      ex.printStackTrace();
    } 
    return null;
  }

  /**
   * All deserializers return this dataflow variable
   */
  public static class UriFuture extends DataFlow<Uri> {

    public UriFuture() {
      super();
    }

    public UriFuture(Uri tupleUri) {
      super(tupleUri);
    }

  }

  /**
   * @see serializeFromProvider with which this method is symmetric.
   */
  public static Uri deserializeToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    logger.debug("deserialize message");

    final UriFuture uri;
    switch (encoding.getType()) {
    case CUSTOM:
      uri = RequestSerializer.deserializeCustomToProvider(context, channelName, provider, encoding, data);
      break;
    case JSON: 
      uri = RequestSerializer.deserializeJsonToProvider(context, channelName, provider, encoding, data);
      break;
    case TERSE: 
      uri = RequestSerializer.deserializeTerseToProvider(context, channelName, provider, encoding, data);
      break;
    default:
      uri = RequestSerializer.deserializeCustomToProvider(context, channelName, provider, encoding, data);  
    }
    if (uri == null) return null;
    try {
      return uri.get();
    } catch (InterruptedException ex) {
      logger.error("interrupted thread ", ex);
      return null;
    } catch (ExecutionException ex) {
      logger.error("execution error thread ", ex);
      return null;
    }
  }


  /**
   * A pair of functions which communicate with a Content Adaptor Service.
   * These Content Adaptor Services are created by the code generator.
   *
   */
  public static ByteBufferFuture serializeCustomFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    final Uri serialUri = Uri.withAppendedPath(tupleUri, encoding.getPayloadSuffix());
    final Cursor tupleCursor;
    try {
      tupleCursor = resolver.query(serialUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider", ex);
      return null;
    }
    if (tupleCursor == null) {
      throw new TupleNotFoundException("while serializing from provider", tupleUri);
    }

    if (! tupleCursor.moveToFirst()) {
      tupleCursor.close();
      return null;
    }
    int columnCount = tupleCursor.getColumnCount();
    if (columnCount < 1) {
      tupleCursor.close();
      return null;
    }
    tupleCursor.moveToFirst();

    final String tupleString = tupleCursor.getString(0);
    tupleCursor.close();
    return ByteBufferFuture.wrap(tupleString.getBytes());
  }

  /**
   * Invoke the custom deserializer.
   * 
   * @param context
   * @param channelName
   * @param provider
   * @param encoding
   * @param data
   * @return
   */
  public static UriFuture deserializeCustomToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    final String key = provider.toString();
    if ( RequestSerializer.remoteServiceMap.containsKey(key)) {
      final IDistributorAdaptor adaptor = RequestSerializer.remoteServiceMap.get(key);
      try {
        final String uriString = adaptor.deserialize(encoding.name(), key, data);
        Uri.parse(uriString);

      } catch (RemoteException ex) {
        ex.printStackTrace();
      }
      return null;
    }
    final ServiceConnection connection = new ServiceConnection() {
      @Override
      public void onServiceConnected(ComponentName name, IBinder service) {

        if (! RequestSerializer.remoteServiceMap.containsKey(key)) {
          RequestSerializer.remoteServiceMap.put(key, IDistributorAdaptor.Stub.asInterface(service));  
        }
        final IDistributorAdaptor adaptor = RequestSerializer.remoteServiceMap.get(key);

        // call the deserialize function here ....
        try {
          adaptor.deserialize(encoding.name(), key, data);
        } catch (RemoteException ex) {
          ex.printStackTrace();
        }
      }

      @Override
      public void onServiceDisconnected(ComponentName name) {
        logger.debug("service disconnected");
        RequestSerializer.remoteServiceMap.remove(key);
      }
    };
    final Intent intent = new Intent();
    context.bindService(intent, connection, Context.BIND_AUTO_CREATE);
    return null;
  }

  /**
   * The JSON serialization is of the following form...
   * serialized tuple : A list of non-null bytes which serialize the tuple, 
   *   this is provided/supplied to the ammo enabled content provider via insert/query.
   *   The serialized tuple may be null terminated or the byte array may simply end.
   * field blobs : A list of name:value pairs where name is the field name and value is 
   *   the field's data blob associated with that field.
   *   There may be multiple field blobs.
   *   
   *   field name : A null terminated name, 
   *   field data length : A 4 byte big-endian length, indicating the number of bytes in the data blob.
   *   field data blob : A set of bytes whose quantity is that of the field data length
   *   
   * Note the serializeFromProvider and serializeFromProvider are symmetric, 
   * any change to one will necessitate a corresponding change to the other.
   */
  public static ByteBufferFuture serializeJsonFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    logger.trace("Serialize the non-blob data");

    final Uri serialUri = Uri.withAppendedPath(tupleUri, encoding.getPayloadSuffix());
    Cursor tupleCursor = null;
    final byte[] tuple;
    final JSONObject json;
    try {
    try {
      tupleCursor = resolver.query(serialUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider {}", ex.getLocalizedMessage());
      return null;
    }
    if (tupleCursor == null) {
      throw new TupleNotFoundException("while serializing from provider", tupleUri);
    }
    if ( tupleCursor.getCount() < 1) {
      logger.warn("tuple no longe present {}", tupleUri);
      return null;
    }
      if (! tupleCursor.moveToFirst()) { return null; }
      if (tupleCursor.getColumnCount() < 1) { return null; }

      json = new JSONObject();
    tupleCursor.moveToFirst();

    for (final String name : tupleCursor.getColumnNames()) {
      if (name.startsWith("_")) continue; // don't send the local fields

      final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
      if (value == null || value.length() < 1) continue;
      try {
        json.put(name, value);
      } catch (JSONException ex) {
        logger.warn("invalid content provider", ex);
      }
      }
    } finally {
      if (tupleCursor != null) tupleCursor.close(); 
    }
    // FIXME FPE final Writer can we be more efficient? not copy bytes so often?
    tuple = json.toString().getBytes();

    logger.info("Serialized message, content {}", json.toString() );

    logger.trace("loading larger tuple buffer");
    final ByteArrayOutputStream bigTuple = new ByteArrayOutputStream();

    bigTuple.write(tuple); 
    bigTuple.write(0x0);

    logger.trace("Serialize the blob data (if any)");

    final Uri blobUri = Uri.withAppendedPath(tupleUri, "_blob");
    Cursor blobCursor = null;
    final int blobCount;
    try {
    try {
      blobCursor = resolver.query(blobUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider", ex);
      return null;
    }
    if (blobCursor == null) {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }
    if (! blobCursor.moveToFirst()) {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }
      blobCount = blobCursor.getColumnCount();
    if (blobCount < 1)  {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }

    logger.trace("getting the blob fields");  
    final byte[] buffer = new byte[1024]; 
    for (int ix=0; ix < blobCursor.getColumnCount(); ix++) {

        final String fieldName = blobCursor.getColumnName(ix);
        bigTuple.write(fieldName.getBytes());
        bigTuple.write(0x0);

        /*
         * An awful hack.
         * If it is a file field type the value is a string,
         * If a blob then the field type is a blob, but
         * you can not check the field type directly so we
         * let the exception happen if we try the wrong type.
         */
        @SuppressWarnings("unused")
		final String fileName;
        final byte[] blob;
        final BlobTypeEnum dataType;
        try {
          String tempFileName = null;
          byte[] tempBlob = null;
          BlobTypeEnum tempDataType = null;
          try {
            tempFileName = blobCursor.getString(ix);
            if (tempFileName == null || tempFileName.length() < 1) {
            	tempDataType = BlobTypeEnum.SMALL;
            } else {
            	tempDataType = BlobTypeEnum.LARGE;
            }
          } catch (Exception ex) {
            tempBlob = blobCursor.getBlob(ix);
            tempDataType = BlobTypeEnum.infer(fieldName, tempBlob);
          }
          fileName = tempFileName;
          blob = tempBlob;
          dataType = tempDataType;
        } catch (Exception ex) {
          logger.error("something bad happened reading the field value", ex);
          continue;
        }

        switch (dataType) {
        case SMALL:
          try {
            logger.trace("field name=[{}] blob=[{}]", fieldName, blob);
           
            final ByteBuffer bb = ByteBuffer.allocate(4);
            bb.order(ByteOrder.BIG_ENDIAN); 
            final int size = (blob == null) ? 0 : blob.length;
            bb.putInt(size);
            bigTuple.write(bb.array());
            if (blob != null) {
            	bigTuple.write(blob);
            }
            bigTuple.write(BLOB_MARKER_FIELD);
            bigTuple.write(bb.array(),1,bb.array().length-1);
          } finally { }
          break;
        case LARGE: 
          logger.trace("field name=[{}] ", fieldName);
          final Uri fieldUri = Uri.withAppendedPath(tupleUri, fieldName);
          try {
            final AssetFileDescriptor afd = resolver.openAssetFileDescriptor(fieldUri, "r");
            if (afd == null) {
              logger.warn("could not acquire file descriptor {}", fieldUri);
              throw new IOException("could not acquire file descriptor "+fieldUri);
            }
            final ParcelFileDescriptor pfd = afd.getParcelFileDescriptor();

            final InputStream instream = new ParcelFileDescriptor.AutoCloseInputStream(pfd);
            final BufferedInputStream bis = new BufferedInputStream(instream);
            final ByteArrayOutputStream fieldBlob = new ByteArrayOutputStream();
            for (int bytesRead = 0; (bytesRead = bis.read(buffer)) != -1;) {
              fieldBlob.write(buffer, 0, bytesRead);
            }
            bis.close();
            final ByteBuffer fieldBlobBuffer = ByteBuffer.wrap(fieldBlob.toByteArray());

            // write it out
            final ByteBuffer bb = ByteBuffer.allocate(4);
            bb.order(ByteOrder.BIG_ENDIAN); 
            final int size = fieldBlobBuffer.capacity();
            bb.putInt(size);
            bigTuple.write(bb.array());

            bigTuple.write(fieldBlobBuffer.array());
            bigTuple.write(bb.array());

          } catch (IOException ex) {
            logger.trace("unable to create stream {}", serialUri, ex);
            throw new FileNotFoundException("Unable to create stream");
          }
          break;
        default:
          logger.warn("not a known data type {}", dataType);
        }
      }
    } finally {
      if (blobCursor != null) blobCursor.close();
    }

    final byte[] finalTuple = bigTuple.toByteArray();
    bigTuple.close();
    PLogger.API_STORE.debug("json tuple=[{}] size=[{}]", 
        tuple, finalTuple.length);
    PLogger.API_STORE.trace("json finalTuple=[{}]", 
        finalTuple);
    return ByteBufferFuture.wrap(finalTuple);
  }

  /**
   * This is the improved version of JSON encoding.
   * It requests the meta data about the relation explicityl rather
   * than trying to infer it from the returned data.
   */
  public static ByteBufferFuture serializeJsonFromProvider2(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    /**
     * 1) query to find out about the fields to send: name, position, type
     * 2) serialize the fields 
     */
    logger.debug("Using json serialization");

    Cursor serialMetaCursor = null;
    final Map<String,FieldType> serialMap;
    final String[] serialOrder;
    try {
      try {
        final Uri dUri = Uri.withAppendedPath(tupleUri, "_data_type");
        serialMetaCursor = resolver.query(dUri, null, null, null, null);
      } catch(IllegalArgumentException ex) {
        logger.warn("unknown content provider ", ex);
        return null;
      }
      if (serialMetaCursor == null) {
        throw new NonConformingAmmoContentProvider("while getting metadata from provider", tupleUri);
      }

      if (! serialMetaCursor.moveToFirst()) { return null; }
      final int columnCount = serialMetaCursor.getColumnCount();
      if (columnCount < 1) { return null; }

      serialMap = new HashMap<String,FieldType>(columnCount);
      serialOrder = new String[columnCount];
      int ix = 0;
      for (final String key : serialMetaCursor.getColumnNames()) {
        if (key.startsWith("_")) continue; // don't send any local fields
        final int value = serialMetaCursor.getInt(serialMetaCursor.getColumnIndex(key));
        serialMap.put(key, FieldType.fromCode(value));
        serialOrder[ix] = key;
        ix++;
      }
    } finally {
      if (serialMetaCursor != null) serialMetaCursor.close(); 
    }

    logger.trace("Serialize the non-binary data");
    Cursor tupleCursor = null;
    final byte[] tuple;
    try {
      try {
        tupleCursor = resolver.query(tupleUri, null, null, null, null);
      } catch(IllegalArgumentException ex) {
        logger.warn("unknown content provider ", ex);
        return null;
      }
      if (tupleCursor == null) {
        throw new TupleNotFoundException("while serializing from provider", tupleUri);
      }

      if (! tupleCursor.moveToFirst()) {
        return null;
      }
      if (tupleCursor.getColumnCount() < 1) {
        logger.warn("tuple no longe present {}", tupleUri);
        return null;
      }

      logger.trace("Serialize the non-blob data");

      final JSONObject json = new JSONObject();
      int countBinaryFields = 0;
      for (final Map.Entry<String,FieldType> entry : serialMap.entrySet()) {
        final String name = entry.getKey();
        final FieldType type = entry.getValue();
        switch(type) {
          case BLOB: 
             countBinaryFields++;
          break;
          case FILE: {
            countBinaryFields++;
            final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
            try {
              json.put(name, value);
            } catch (JSONException ex) {
              logger.warn("invalid content provider", ex);
            }
          }
          break;
          default: {
            final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
            if (value == null || value.length() < 1) continue;
            try {
              json.put(name, value);
            } catch (JSONException ex) {
              logger.warn("invalid content provider", ex);
            }
          }
        }
      }
      tuple = json.toString().getBytes();
      if (countBinaryFields < 1) return ByteBufferFuture.wrap(tuple);

      logger.trace("loading larger tuple buffer");
      ByteArrayOutputStream bigTuple = null;
      try {
        bigTuple = new ByteArrayOutputStream();

        bigTuple.write(tuple); 
        bigTuple.write(0x0);

        logger.trace("Serialize the blob data (if any)");
        final byte[] buffer = new byte[1024]; 
        for (final Map.Entry<String,FieldType> entry : serialMap.entrySet()) {
          final String fieldName = entry.getKey();
          final FieldType type = entry.getValue();
          switch(type) {
            case BLOB: 
            case FILE:
            break;
            default:
              continue;
          }
          bigTuple.write(fieldName.getBytes());
          bigTuple.write(0x0);

          switch(type) {
            case BLOB: 
              final byte[] tempBlob = tupleCursor.getBlob(tupleCursor.getColumnIndex(entry.getKey()));
              final BlobTypeEnum tempBlobType = BlobTypeEnum.infer(fieldName, tempBlob);
              switch (tempBlobType) {
                case SMALL:
                try {
                  logger.trace("field name=[{}] blob=[{}]", fieldName, tempBlob);
                  final ByteBuffer fieldBlobBuffer = ByteBuffer.wrap(tempBlob);
  
                  final ByteBuffer bb = ByteBuffer.allocate(4);
                  bb.order(ByteOrder.BIG_ENDIAN); 
                  final int size = fieldBlobBuffer.capacity();
                  bb.putInt(size);
                  bigTuple.write(bb.array());
      
                  bigTuple.write(fieldBlobBuffer.array());
  
                  bigTuple.write(BLOB_MARKER_FIELD);
                  bigTuple.write(bb.array(),1,bb.array().length-1);
                } finally { }
                break;
			case LARGE:
			default:
				break;
              }
            continue;
		  default:
			break;
          }

          logger.trace("field name=[{}] ", fieldName);
          final Uri fieldUri = Uri.withAppendedPath(tupleUri, fieldName);
          try {
              final AssetFileDescriptor afd = resolver.openAssetFileDescriptor(fieldUri, "r");
              if (afd == null) {
                logger.warn("could not acquire file descriptor {}", fieldUri);
                throw new IOException("could not acquire file descriptor "+fieldUri);
              }
              final ParcelFileDescriptor pfd = afd.getParcelFileDescriptor();
  
              final InputStream instream = new ParcelFileDescriptor.AutoCloseInputStream(pfd);
              final ByteBuffer fieldBlobBuffer;
              ByteArrayOutputStream fieldBlob = null;
              BufferedInputStream bis = null;
              try {
                bis = new BufferedInputStream(instream);
                fieldBlob = new ByteArrayOutputStream();
                for (int bytesRead = 0; (bytesRead = bis.read(buffer)) != -1;) {
                  fieldBlob.write(buffer, 0, bytesRead);
                }
                fieldBlobBuffer = ByteBuffer.wrap(fieldBlob.toByteArray());
              } finally {
                if (bis != null) bis.close();
                if (fieldBlob != null) fieldBlob.close();
              }
  
              // write it out
              final ByteBuffer bb = ByteBuffer.allocate(4);
              bb.order(ByteOrder.BIG_ENDIAN); 
              final int size = fieldBlobBuffer.capacity();
              bb.putInt(size);
              bigTuple.write(bb.array());
  
              bigTuple.write(fieldBlobBuffer.array());
              bigTuple.write(bb.array());
  
          } catch (IOException ex) {
              logger.trace("unable to create stream {}", tupleUri, ex);
              throw new FileNotFoundException("Unable to create stream");
          }

        } 
        final byte[] finalTuple = bigTuple.toByteArray();
        bigTuple.close();
        PLogger.API_STORE.debug("json tuple=[{}] size=[{}]", 
            tuple, finalTuple.length);
          PLogger.API_STORE.trace("json finalTuple=[{}]", 
            finalTuple);
        return ByteBufferFuture.wrap(finalTuple);

      } finally {
        if (bigTuple != null) bigTuple.close(); 
      }

    } finally {
      if (tupleCursor != null) tupleCursor.close(); 
    }
  }


  /**
   * JSON encoding (deprecated)
   * This method interacts directly with the content provider.
   * It should only be used with content providers which are known
   * to be responsive.
   * 
   * @param context
   * @param provider
   * @param encoding
   * @param data
   * @return
   */
  public static UriFuture deserializeJsonToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    final ContentResolver resolver = context.getContentResolver();
    final ByteBuffer dataBuff = ByteBuffer.wrap(data).order(ByteOrder.BIG_ENDIAN);
    // find the end of the json portion of the data
    int position = 0;
    for (; position < data.length && data[position] != (byte)0x0; position++) {}

    final int length = position;
    final byte[] payload = new byte[length];
    System.arraycopy(data, 0, payload, 0, length);
    final JSONObject input;
    try {
      final String parsePayload = new String(payload);
      final Object value = new JSONTokener(parsePayload).nextValue();
      if (value instanceof JSONObject) {
        input = (JSONObject) value;
      } else if (value instanceof JSONArray) {
        logger.warn("invalid JSON payload=[{}]", parsePayload);
        return null;
      } else if (value == JSONObject.NULL) {
        logger.warn("null JSON payload=[{}]", parsePayload);
        return null;
      } else {
        logger.warn("{} JSON payload=[{}]", value.getClass().getName(), parsePayload);
        return null;<commit_msg>added a log message for the jason payload received. cause 'ant realclean' to not remove the 'dist/pom.xml'<commit_after>/*Copyright (C) 2010-2012 Institute for Software Integrated Systems (ISIS)
This software was developed by the Institute for Software Integrated
Systems (ISIS) at Vanderbilt University, Tennessee, USA for the 
Transformative Apps program under DARPA, Contract # HR011-10-C-0175.
The United States Government has unlimited rights to this software. 
The US government has the right to use, modify, reproduce, release, 
perform, display, or disclose computer software or computer software 
documentation in whole or in part, in any manner and for any 
purpose whatsoever, and to have or authorize others to do so.
 */
package edu.vu.isis.ammo.core.distributor;

import java.io.BufferedInputStream;
import java.io.ByteArrayOutputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;
import org.json.JSONTokener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import android.content.ComponentName;
import android.content.ContentResolver;
import android.content.ContentUris;
import android.content.ContentValues;
import android.content.Context;
import android.content.Intent;
import android.content.ServiceConnection;
import android.content.res.AssetFileDescriptor;
import android.database.Cursor;
import android.database.sqlite.SQLiteException;
import android.net.Uri;
import android.os.IBinder;
import android.os.ParcelFileDescriptor;
import android.os.RemoteException;
import edu.vu.isis.ammo.api.IDistributorAdaptor;
import edu.vu.isis.ammo.api.type.Payload;
import edu.vu.isis.ammo.api.type.Provider;
import edu.vu.isis.ammo.core.AmmoService;
import edu.vu.isis.ammo.core.PLogger;
import edu.vu.isis.ammo.core.distributor.DistributorDataStore.DisposalState;
import edu.vu.isis.ammo.core.distributor.DistributorPolicy.Encoding;
import edu.vu.isis.ammo.core.network.AmmoGatewayMessage;

/**
 * The purpose of these objects is lazily serialize an object.
 * Once it has been serialized once a copy is kept.
 *
 */
public class RequestSerializer {
  private static final Logger logger = LoggerFactory.getLogger("dist.serializer");

  /**
   * This enumeration's codes must match those of 
   * the AmmoGen files.
   */
  public enum FieldType {
    NULL(0), 
    BOOL(1), 
    BLOB(2), 
    FLOAT(3),
    INTEGER(4), 
    LONG(5), 
    TEXT(6), 
    REAL(7),
    FK(8), 
    GUID(9), 
    EXCLUSIVE(10), 
    INCLUSIVE(11),
    TIMESTAMP(12),
    SHORT(13),
    FILE(14);

    private final int code;

    private FieldType(int code) {
      this.code = code;
    }

    private static final Map<Integer,FieldType> codemap = 
        new HashMap<Integer,FieldType>();
    static {
      for (FieldType t : FieldType.values()) {
        FieldType.codemap.put(t.code, t);
      }
    }

    public int toCode() {
      return this.code;
    }

    public static FieldType fromCode(final int code) {
      return FieldType.codemap.get(code);
    }
  }

  /**
   * The presence of the BLOB_MARKER_FIELD as the first byte in the 
   * footer for a blob data section indicates where the blob should 
   * be placed in the content provider.
   */
  public static final byte BLOB_MARKER_FIELD = (byte)0xff;

  private enum BlobTypeEnum { 
        LARGE, SMALL; 

        /**
         * The difficulty here is that the blob may have trailing null bytes.
         * e.g.
         *  fieldName = [data], fieldNameBlob= [[100, 97, 116, 97]], blob = [[100, 97, 116, 97, 0]]
         * These should result it a match.
         */
        public static BlobTypeEnum infer(String fieldName, byte[] blob) {
       
           final byte[] fieldNameBlob; 
          try {
              fieldNameBlob = fieldName.getBytes("UTF-8");
          } catch (java.io.UnsupportedEncodingException ex) {
              return SMALL;
          }

         logger.trace("processing blob fieldName = [{}], fieldNameBlob= [{}], blob = [{}]", 
             new Object[] {fieldName, fieldNameBlob, blob});

          if (blob == null) return LARGE;
          if (blob.length < 1) return LARGE;

          if (fieldNameBlob.length == blob.length)
              return Arrays.equals(blob, fieldNameBlob) ? LARGE : SMALL;

          if (fieldNameBlob.length > blob.length) return SMALL;

          int i;
          for (i = 0; i < fieldNameBlob.length; i++) {
             if (fieldNameBlob[i] != blob[i]) return SMALL;
          }
          for (; i < blob.length; i++) {
               if (blob[i] != (byte)0) return SMALL;
          }
          return LARGE;
    }
  }
  
  public interface OnReady  {
    public AmmoGatewayMessage run(Encoding encode, byte[] serialized);
  }
  public interface OnSerialize  {
    public byte[] run(Encoding encode);
  }

  public final Provider provider;
  public final Payload payload;
  private OnReady readyActor;
  private OnSerialize serializeActor;
  private AmmoGatewayMessage agm;

  /**
   * This maintains a set of persistent connections to 
   * content provider adapter services.
   */
  final static private Map<String,IDistributorAdaptor> remoteServiceMap;
  static {
    remoteServiceMap = new HashMap<String,IDistributorAdaptor>(10);
  }

  private RequestSerializer(Provider provider, Payload payload) {
    this.provider = provider;
    this.payload = payload;
    this.agm = null;


    this.readyActor = new RequestSerializer.OnReady() {
      @Override
      public AmmoGatewayMessage run(Encoding encode, byte[] serialized) {
        logger.trace("ready actor not defined {}", encode);
        return null;
      }
    };  
    this.serializeActor = new RequestSerializer.OnSerialize() {
      @Override
      public byte[] run(Encoding encode) {
        logger.trace("serialize actor not defined {}", encode);
        return null;
      }
    };
  }

  public static RequestSerializer newInstance() {
    return new RequestSerializer(null, null);
  }
  public static RequestSerializer newInstance(Provider provider, Payload payload) {
    return new RequestSerializer(provider, payload);
  }

  public DisposalState act(final AmmoService that, final Encoding encode, final String channel) {
    final RequestSerializer parent = RequestSerializer.this;
    final Encoding local_encode = encode;
    final String local_channel = channel;
    if (parent.agm == null) {
      final byte[] agmBytes = parent.serializeActor.run(local_encode);
      parent.agm = parent.readyActor.run(local_encode, agmBytes);
    }
    if (parent.agm == null)
      return null;

    return that.sendRequest(parent.agm, local_channel);
  }

  public void setAction(OnReady action) {
    this.readyActor = action;
  }

  public void setSerializer(OnSerialize onSerialize) {
    this.serializeActor = onSerialize;
  }


  public static byte[] serializeFromContentValues(ContentValues cv, final DistributorPolicy.Encoding encoding) {

    logger.trace("serializing using content values and encoding {}", encoding);
    switch (encoding.getType()) {
    case JSON: 
    {
      return encodeAsJson (cv);
    }

    case TERSE: 
    {
      // Need to be implemented ...
    }
    // TODO custom still needs a lot of work
    // It will presume the presence of a SyncAdaptor for the content provider.
    case CUSTOM:
    default:
    {
    }
    }
    return null;
  }

  private static byte[] encodeAsJson (ContentValues cv) {
    // encoding in json for now ...
    Set<Map.Entry<String, Object>> data = cv.valueSet();
    Iterator<Map.Entry<String, Object>> iter = data.iterator();       
    final JSONObject json = new JSONObject();

    while (iter.hasNext())
    {
      Map.Entry<String, Object> entry = (Map.Entry<String, Object>)iter.next();         
      try {
        if (entry.getValue() instanceof String)
          json.put(entry.getKey(), cv.getAsString(entry.getKey()));
        else if (entry.getValue() instanceof Integer)
          json.put(entry.getKey(), cv.getAsInteger(entry.getKey()));
      } catch (JSONException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
        return null;
      }
    }

    return json.toString().getBytes();
  }

  /**
   * This simple Future which can only be bound to a value once.
   */
  public static class DataFlow<V> implements Future<V> {
    final AtomicReference<V> value;

    public DataFlow() {
      this.value = new  AtomicReference<V>(); 
    }

    public DataFlow(V value) {
      this();
      this.value.set(value);
    }

    public void bind(final V value) {
      if (! this.value.compareAndSet(null, value)) return;
      this.notifyAll();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
      return false;
    }

    @Override
    public V get() throws InterruptedException, ExecutionException {
      synchronized (this) {
        if (this.value.get() == null) {
          this.wait();
        }
      }
      return this.value.get();
    }

    @Override
    public V get(long timeout, TimeUnit unit) 
        throws InterruptedException, ExecutionException, TimeoutException {
      synchronized (this) {
        if (this.value.get() == null) {
          this.wait(unit.toMillis(timeout));
        }
      }
      return this.value.get();
    }

    @Override
    public boolean isCancelled() {
      return false;
    }

    @Override
    public boolean isDone() {
      return this.value.get() != null;
    }
  }


  /**
   * All serializer methods return this dataflow variable.
   */
  public static class ByteBufferFuture extends DataFlow<ByteBuffer> {

    public ByteBufferFuture() {
      super();
    }

    public ByteBufferFuture(ByteBuffer value) {
      super(value);
    }

    public static ByteBufferFuture wrap(byte[] value) {
      return new ByteBufferFuture(ByteBuffer.wrap(value));
    }
  }
  /**
   *
   */  

  public static byte[] serializeFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    logger.trace("serializing using encoding {}", encoding);
    final ByteBufferFuture result;
    final Encoding.Type encodingType = encoding.getType();
    switch (encodingType) {
    case CUSTOM:
      result = RequestSerializer.serializeCustomFromProvider(resolver, tupleUri, encoding);
      break;
    case JSON: 
      result = RequestSerializer.serializeJsonFromProvider(resolver, tupleUri, encoding);
      break;
    case TERSE: 
      result = RequestSerializer.serializeTerseFromProvider(resolver, tupleUri, encoding);
      break;
    default:
      result = RequestSerializer.serializeCustomFromProvider(resolver, tupleUri, encoding);
    }
    try {
      return result.get().array();
    } catch (InterruptedException ex) {
      ex.printStackTrace();
    } catch (ExecutionException ex) {
      ex.printStackTrace();
    } 
    return null;
  }

  /**
   * All deserializers return this dataflow variable
   */
  public static class UriFuture extends DataFlow<Uri> {

    public UriFuture() {
      super();
    }

    public UriFuture(Uri tupleUri) {
      super(tupleUri);
    }

  }

  /**
   * @see serializeFromProvider with which this method is symmetric.
   */
  public static Uri deserializeToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    logger.debug("deserialize message");

    final UriFuture uri;
    switch (encoding.getType()) {
    case CUSTOM:
      uri = RequestSerializer.deserializeCustomToProvider(context, channelName, provider, encoding, data);
      break;
    case JSON: 
      uri = RequestSerializer.deserializeJsonToProvider(context, channelName, provider, encoding, data);
      break;
    case TERSE: 
      uri = RequestSerializer.deserializeTerseToProvider(context, channelName, provider, encoding, data);
      break;
    default:
      uri = RequestSerializer.deserializeCustomToProvider(context, channelName, provider, encoding, data);  
    }
    if (uri == null) return null;
    try {
      return uri.get();
    } catch (InterruptedException ex) {
      logger.error("interrupted thread ", ex);
      return null;
    } catch (ExecutionException ex) {
      logger.error("execution error thread ", ex);
      return null;
    }
  }


  /**
   * A pair of functions which communicate with a Content Adaptor Service.
   * These Content Adaptor Services are created by the code generator.
   *
   */
  public static ByteBufferFuture serializeCustomFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    final Uri serialUri = Uri.withAppendedPath(tupleUri, encoding.getPayloadSuffix());
    final Cursor tupleCursor;
    try {
      tupleCursor = resolver.query(serialUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider", ex);
      return null;
    }
    if (tupleCursor == null) {
      throw new TupleNotFoundException("while serializing from provider", tupleUri);
    }

    if (! tupleCursor.moveToFirst()) {
      tupleCursor.close();
      return null;
    }
    int columnCount = tupleCursor.getColumnCount();
    if (columnCount < 1) {
      tupleCursor.close();
      return null;
    }
    tupleCursor.moveToFirst();

    final String tupleString = tupleCursor.getString(0);
    tupleCursor.close();
    return ByteBufferFuture.wrap(tupleString.getBytes());
  }

  /**
   * Invoke the custom deserializer.
   * 
   * @param context
   * @param channelName
   * @param provider
   * @param encoding
   * @param data
   * @return
   */
  public static UriFuture deserializeCustomToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    final String key = provider.toString();
    if ( RequestSerializer.remoteServiceMap.containsKey(key)) {
      final IDistributorAdaptor adaptor = RequestSerializer.remoteServiceMap.get(key);
      try {
        final String uriString = adaptor.deserialize(encoding.name(), key, data);
        Uri.parse(uriString);

      } catch (RemoteException ex) {
        ex.printStackTrace();
      }
      return null;
    }
    final ServiceConnection connection = new ServiceConnection() {
      @Override
      public void onServiceConnected(ComponentName name, IBinder service) {

        if (! RequestSerializer.remoteServiceMap.containsKey(key)) {
          RequestSerializer.remoteServiceMap.put(key, IDistributorAdaptor.Stub.asInterface(service));  
        }
        final IDistributorAdaptor adaptor = RequestSerializer.remoteServiceMap.get(key);

        // call the deserialize function here ....
        try {
          adaptor.deserialize(encoding.name(), key, data);
        } catch (RemoteException ex) {
          ex.printStackTrace();
        }
      }

      @Override
      public void onServiceDisconnected(ComponentName name) {
        logger.debug("service disconnected");
        RequestSerializer.remoteServiceMap.remove(key);
      }
    };
    final Intent intent = new Intent();
    context.bindService(intent, connection, Context.BIND_AUTO_CREATE);
    return null;
  }

  /**
   * The JSON serialization is of the following form...
   * serialized tuple : A list of non-null bytes which serialize the tuple, 
   *   this is provided/supplied to the ammo enabled content provider via insert/query.
   *   The serialized tuple may be null terminated or the byte array may simply end.
   * field blobs : A list of name:value pairs where name is the field name and value is 
   *   the field's data blob associated with that field.
   *   There may be multiple field blobs.
   *   
   *   field name : A null terminated name, 
   *   field data length : A 4 byte big-endian length, indicating the number of bytes in the data blob.
   *   field data blob : A set of bytes whose quantity is that of the field data length
   *   
   * Note the serializeFromProvider and serializeFromProvider are symmetric, 
   * any change to one will necessitate a corresponding change to the other.
   */
  public static ByteBufferFuture serializeJsonFromProvider(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    logger.trace("Serialize the non-blob data");

    final Uri serialUri = Uri.withAppendedPath(tupleUri, encoding.getPayloadSuffix());
    Cursor tupleCursor = null;
    final byte[] tuple;
    final JSONObject json;
    try {
    try {
      tupleCursor = resolver.query(serialUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider {}", ex.getLocalizedMessage());
      return null;
    }
    if (tupleCursor == null) {
      throw new TupleNotFoundException("while serializing from provider", tupleUri);
    }
    if ( tupleCursor.getCount() < 1) {
      logger.warn("tuple no longe present {}", tupleUri);
      return null;
    }
      if (! tupleCursor.moveToFirst()) { return null; }
      if (tupleCursor.getColumnCount() < 1) { return null; }

      json = new JSONObject();
    tupleCursor.moveToFirst();

    for (final String name : tupleCursor.getColumnNames()) {
      if (name.startsWith("_")) continue; // don't send the local fields

      final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
      if (value == null || value.length() < 1) continue;
      try {
        json.put(name, value);
      } catch (JSONException ex) {
        logger.warn("invalid content provider", ex);
      }
      }
    } finally {
      if (tupleCursor != null) tupleCursor.close(); 
    }
    // FIXME FPE final Writer can we be more efficient? not copy bytes so often?
    tuple = json.toString().getBytes();

    logger.info("Serialized message, content {}", json.toString() );

    logger.trace("loading larger tuple buffer");
    final ByteArrayOutputStream bigTuple = new ByteArrayOutputStream();

    bigTuple.write(tuple); 
    bigTuple.write(0x0);

    logger.trace("Serialize the blob data (if any)");

    final Uri blobUri = Uri.withAppendedPath(tupleUri, "_blob");
    Cursor blobCursor = null;
    final int blobCount;
    try {
    try {
      blobCursor = resolver.query(blobUri, null, null, null, null);
    } catch(IllegalArgumentException ex) {
      logger.warn("unknown content provider", ex);
      return null;
    }
    if (blobCursor == null) {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }
    if (! blobCursor.moveToFirst()) {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }
      blobCount = blobCursor.getColumnCount();
    if (blobCount < 1)  {
      PLogger.API_STORE.debug("json tuple=[{}]", tuple);
      return ByteBufferFuture.wrap(tuple);
    }

    logger.trace("getting the blob fields");  
    final byte[] buffer = new byte[1024]; 
    for (int ix=0; ix < blobCursor.getColumnCount(); ix++) {

        final String fieldName = blobCursor.getColumnName(ix);
        bigTuple.write(fieldName.getBytes());
        bigTuple.write(0x0);

        /*
         * An awful hack.
         * If it is a file field type the value is a string,
         * If a blob then the field type is a blob, but
         * you can not check the field type directly so we
         * let the exception happen if we try the wrong type.
         */
        @SuppressWarnings("unused")
		final String fileName;
        final byte[] blob;
        final BlobTypeEnum dataType;
        try {
          String tempFileName = null;
          byte[] tempBlob = null;
          BlobTypeEnum tempDataType = null;
          try {
            tempFileName = blobCursor.getString(ix);
            if (tempFileName == null || tempFileName.length() < 1) {
            	tempDataType = BlobTypeEnum.SMALL;
            } else {
            	tempDataType = BlobTypeEnum.LARGE;
            }
          } catch (Exception ex) {
            tempBlob = blobCursor.getBlob(ix);
            tempDataType = BlobTypeEnum.infer(fieldName, tempBlob);
          }
          fileName = tempFileName;
          blob = tempBlob;
          dataType = tempDataType;
        } catch (Exception ex) {
          logger.error("something bad happened reading the field value", ex);
          continue;
        }

        switch (dataType) {
        case SMALL:
          try {
            logger.trace("field name=[{}] blob=[{}]", fieldName, blob);
           
            final ByteBuffer bb = ByteBuffer.allocate(4);
            bb.order(ByteOrder.BIG_ENDIAN); 
            final int size = (blob == null) ? 0 : blob.length;
            bb.putInt(size);
            bigTuple.write(bb.array());
            if (blob != null) {
            	bigTuple.write(blob);
            }
            bigTuple.write(BLOB_MARKER_FIELD);
            bigTuple.write(bb.array(),1,bb.array().length-1);
          } finally { }
          break;
        case LARGE: 
          logger.trace("field name=[{}] ", fieldName);
          final Uri fieldUri = Uri.withAppendedPath(tupleUri, fieldName);
          try {
            final AssetFileDescriptor afd = resolver.openAssetFileDescriptor(fieldUri, "r");
            if (afd == null) {
              logger.warn("could not acquire file descriptor {}", fieldUri);
              throw new IOException("could not acquire file descriptor "+fieldUri);
            }
            final ParcelFileDescriptor pfd = afd.getParcelFileDescriptor();

            final InputStream instream = new ParcelFileDescriptor.AutoCloseInputStream(pfd);
            final BufferedInputStream bis = new BufferedInputStream(instream);
            final ByteArrayOutputStream fieldBlob = new ByteArrayOutputStream();
            for (int bytesRead = 0; (bytesRead = bis.read(buffer)) != -1;) {
              fieldBlob.write(buffer, 0, bytesRead);
            }
            bis.close();
            final ByteBuffer fieldBlobBuffer = ByteBuffer.wrap(fieldBlob.toByteArray());

            // write it out
            final ByteBuffer bb = ByteBuffer.allocate(4);
            bb.order(ByteOrder.BIG_ENDIAN); 
            final int size = fieldBlobBuffer.capacity();
            bb.putInt(size);
            bigTuple.write(bb.array());

            bigTuple.write(fieldBlobBuffer.array());
            bigTuple.write(bb.array());

          } catch (IOException ex) {
            logger.trace("unable to create stream {}", serialUri, ex);
            throw new FileNotFoundException("Unable to create stream");
          }
          break;
        default:
          logger.warn("not a known data type {}", dataType);
        }
      }
    } finally {
      if (blobCursor != null) blobCursor.close();
    }

    final byte[] finalTuple = bigTuple.toByteArray();
    bigTuple.close();
    PLogger.API_STORE.debug("json tuple=[{}] size=[{}]", 
        tuple, finalTuple.length);
    PLogger.API_STORE.trace("json finalTuple=[{}]", 
        finalTuple);
    return ByteBufferFuture.wrap(finalTuple);
  }

  /**
   * This is the improved version of JSON encoding.
   * It requests the meta data about the relation explicityl rather
   * than trying to infer it from the returned data.
   */
  public static ByteBufferFuture serializeJsonFromProvider2(final ContentResolver resolver, 
      final Uri tupleUri, final DistributorPolicy.Encoding encoding) 
          throws TupleNotFoundException, NonConformingAmmoContentProvider, IOException {

    /**
     * 1) query to find out about the fields to send: name, position, type
     * 2) serialize the fields 
     */
    logger.debug("Using json serialization");

    Cursor serialMetaCursor = null;
    final Map<String,FieldType> serialMap;
    final String[] serialOrder;
    try {
      try {
        final Uri dUri = Uri.withAppendedPath(tupleUri, "_data_type");
        serialMetaCursor = resolver.query(dUri, null, null, null, null);
      } catch(IllegalArgumentException ex) {
        logger.warn("unknown content provider ", ex);
        return null;
      }
      if (serialMetaCursor == null) {
        throw new NonConformingAmmoContentProvider("while getting metadata from provider", tupleUri);
      }

      if (! serialMetaCursor.moveToFirst()) { return null; }
      final int columnCount = serialMetaCursor.getColumnCount();
      if (columnCount < 1) { return null; }

      serialMap = new HashMap<String,FieldType>(columnCount);
      serialOrder = new String[columnCount];
      int ix = 0;
      for (final String key : serialMetaCursor.getColumnNames()) {
        if (key.startsWith("_")) continue; // don't send any local fields
        final int value = serialMetaCursor.getInt(serialMetaCursor.getColumnIndex(key));
        serialMap.put(key, FieldType.fromCode(value));
        serialOrder[ix] = key;
        ix++;
      }
    } finally {
      if (serialMetaCursor != null) serialMetaCursor.close(); 
    }

    logger.trace("Serialize the non-binary data");
    Cursor tupleCursor = null;
    final byte[] tuple;
    try {
      try {
        tupleCursor = resolver.query(tupleUri, null, null, null, null);
      } catch(IllegalArgumentException ex) {
        logger.warn("unknown content provider ", ex);
        return null;
      }
      if (tupleCursor == null) {
        throw new TupleNotFoundException("while serializing from provider", tupleUri);
      }

      if (! tupleCursor.moveToFirst()) {
        return null;
      }
      if (tupleCursor.getColumnCount() < 1) {
        logger.warn("tuple no longe present {}", tupleUri);
        return null;
      }

      logger.trace("Serialize the non-blob data");

      final JSONObject json = new JSONObject();
      int countBinaryFields = 0;
      for (final Map.Entry<String,FieldType> entry : serialMap.entrySet()) {
        final String name = entry.getKey();
        final FieldType type = entry.getValue();
        switch(type) {
          case BLOB: 
             countBinaryFields++;
          break;
          case FILE: {
            countBinaryFields++;
            final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
            try {
              json.put(name, value);
            } catch (JSONException ex) {
              logger.warn("invalid content provider", ex);
            }
          }
          break;
          default: {
            final String value = tupleCursor.getString(tupleCursor.getColumnIndex(name));
            if (value == null || value.length() < 1) continue;
            try {
              json.put(name, value);
            } catch (JSONException ex) {
              logger.warn("invalid content provider", ex);
            }
          }
        }
      }
      tuple = json.toString().getBytes();
      if (countBinaryFields < 1) return ByteBufferFuture.wrap(tuple);

      logger.trace("loading larger tuple buffer");
      ByteArrayOutputStream bigTuple = null;
      try {
        bigTuple = new ByteArrayOutputStream();

        bigTuple.write(tuple); 
        bigTuple.write(0x0);

        logger.trace("Serialize the blob data (if any)");
        final byte[] buffer = new byte[1024]; 
        for (final Map.Entry<String,FieldType> entry : serialMap.entrySet()) {
          final String fieldName = entry.getKey();
          final FieldType type = entry.getValue();
          switch(type) {
            case BLOB: 
            case FILE:
            break;
            default:
              continue;
          }
          bigTuple.write(fieldName.getBytes());
          bigTuple.write(0x0);

          switch(type) {
            case BLOB: 
              final byte[] tempBlob = tupleCursor.getBlob(tupleCursor.getColumnIndex(entry.getKey()));
              final BlobTypeEnum tempBlobType = BlobTypeEnum.infer(fieldName, tempBlob);
              switch (tempBlobType) {
                case SMALL:
                try {
                  logger.trace("field name=[{}] blob=[{}]", fieldName, tempBlob);
                  final ByteBuffer fieldBlobBuffer = ByteBuffer.wrap(tempBlob);
  
                  final ByteBuffer bb = ByteBuffer.allocate(4);
                  bb.order(ByteOrder.BIG_ENDIAN); 
                  final int size = fieldBlobBuffer.capacity();
                  bb.putInt(size);
                  bigTuple.write(bb.array());
      
                  bigTuple.write(fieldBlobBuffer.array());
  
                  bigTuple.write(BLOB_MARKER_FIELD);
                  bigTuple.write(bb.array(),1,bb.array().length-1);
                } finally { }
                break;
			case LARGE:
			default:
				break;
              }
            continue;
		  default:
			break;
          }

          logger.trace("field name=[{}] ", fieldName);
          final Uri fieldUri = Uri.withAppendedPath(tupleUri, fieldName);
          try {
              final AssetFileDescriptor afd = resolver.openAssetFileDescriptor(fieldUri, "r");
              if (afd == null) {
                logger.warn("could not acquire file descriptor {}", fieldUri);
                throw new IOException("could not acquire file descriptor "+fieldUri);
              }
              final ParcelFileDescriptor pfd = afd.getParcelFileDescriptor();
  
              final InputStream instream = new ParcelFileDescriptor.AutoCloseInputStream(pfd);
              final ByteBuffer fieldBlobBuffer;
              ByteArrayOutputStream fieldBlob = null;
              BufferedInputStream bis = null;
              try {
                bis = new BufferedInputStream(instream);
                fieldBlob = new ByteArrayOutputStream();
                for (int bytesRead = 0; (bytesRead = bis.read(buffer)) != -1;) {
                  fieldBlob.write(buffer, 0, bytesRead);
                }
                fieldBlobBuffer = ByteBuffer.wrap(fieldBlob.toByteArray());
              } finally {
                if (bis != null) bis.close();
                if (fieldBlob != null) fieldBlob.close();
              }
  
              // write it out
              final ByteBuffer bb = ByteBuffer.allocate(4);
              bb.order(ByteOrder.BIG_ENDIAN); 
              final int size = fieldBlobBuffer.capacity();
              bb.putInt(size);
              bigTuple.write(bb.array());
  
              bigTuple.write(fieldBlobBuffer.array());
              bigTuple.write(bb.array());
  
          } catch (IOException ex) {
              logger.trace("unable to create stream {}", tupleUri, ex);
              throw new FileNotFoundException("Unable to create stream");
          }

        } 
        final byte[] finalTuple = bigTuple.toByteArray();
        bigTuple.close();
        PLogger.API_STORE.debug("json tuple=[{}] size=[{}]", 
            tuple, finalTuple.length);
          PLogger.API_STORE.trace("json finalTuple=[{}]", 
            finalTuple);
        return ByteBufferFuture.wrap(finalTuple);

      } finally {
        if (bigTuple != null) bigTuple.close(); 
      }

    } finally {
      if (tupleCursor != null) tupleCursor.close(); 
    }
  }


  /**
   * JSON encoding (deprecated)
   * This method interacts directly with the content provider.
   * It should only be used with content providers which are known
   * to be responsive.
   * 
   * @param context
   * @param provider
   * @param encoding
   * @param data
   * @return
   */
  public static UriFuture deserializeJsonToProvider(final Context context, final String channelName,
      final Uri provider, final Encoding encoding, final byte[] data) {

    final ContentResolver resolver = context.getContentResolver();
    final ByteBuffer dataBuff = ByteBuffer.wrap(data).order(ByteOrder.BIG_ENDIAN);
    // find the end of the json portion of the data
    int position = 0;
    for (; position < data.length && data[position] != (byte)0x0; position++) {}

    final int length = position;
    final byte[] payload = new byte[length];
    System.arraycopy(data, 0, payload, 0, length);
    final JSONObject input;
    try {
      final String parsePayload = new String(payload);
      final Object value = new JSONTokener(parsePayload).nextValue();
      if (value instanceof JSONObject) {
        input = (JSONObject) value;
        logger.trace("JSON payload=[{}]", value);
      } else if (value instanceof JSONArray) {
        logger.warn("invalid JSON payload=[{}]", parsePayload);
        return null;
      } else if (value == JSONObject.NULL) {
        logger.warn("null JSON payload=[{}]", parsePayload);
        return null;
      } else {
        logger.warn("{} JSON payload=[{}]", value.getClass().getName(), parsePayload);
        return null;
{'fraction_non_alphanumeric': 0.10214318970307736, 'fraction_numerical': 0.0036345613745250287, 'mean_word_length': 2.528937884248463, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 45, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1089376', 'n_tokens_mistral': 18516, 'n_tokens_neox': 17877, 'n_words': 4976}