Entregar cambios en lotes (SQL Server)

Artículo
10/17/2009

En este tema se describe cómo se entregan los cambios en lotes durante la sincronización de bases de datos en Sync Framework que usa SqlSyncProvider, SqlCeSyncProvider o DbSyncProvider. El código de este tema se centra en las clases de Sync Framework siguientes:

Introducción al procesamiento por lotes

De forma predeterminada, Sync Framework entrega los cambios a cada nodo en un único objeto DataSet. Este objeto se conserva en la memoria cuando los cambios se aplican a un nodo. El comportamiento predeterminado funciona bien si hay memoria suficiente en el equipo en el que se aplican los cambios y la conexión con el equipo es confiable. En algunas aplicaciones, sin embargo, los cambios pueden dividirse en lotes. Imagínese el escenario siguiente de una aplicación de sincronización:

Un gran número de clientes que utilizan SqlCeSyncProvider se sincronizan periódicamente con un servidor que utiliza SqlSyncProvider.
Cada cliente tiene una cantidad de memoria y de espacio en disco limitada.
Las conexiones entre el servidor y los clientes tienen un ancho banda bajo e intermitente, lo que hace que a menudo la sincronización tarde bastante y se interrumpa la conexión.
El tamaño de los cambios (en KB) es grande en una sesión de sincronización normal.

En este tipo de escenario, el procesamiento por lotes de los cambios resulta ideal, pues presenta las siguientes características:

Permite al desarrollador controlar la cantidad de memoria (el tamaño de caché de datos de memoria) que se utiliza para almacenar los cambios del cliente. De este modo, pueden eliminarse los errores de memoria insuficiente del cliente.
Permite que Sync Framework reinicie una operación de sincronización que no se ha realizado correctamente desde el comienzo del lote actual y no desde el comienzo del conjunto completo de cambios.
Puede aminorar o evitar por completo la necesidad de volver a descargar o enumerar los cambios en el servidor si se producen errores en las operaciones.

La configuración del procesamiento por lotes es sencilla en aplicaciones de dos niveles y n niveles, y se puede utilizar tanto en la sesión de sincronización inicial como en las sesiones posteriores.

Configurar y usar el procesamiento por lotes

El procesamiento por lotes en Sync Framework funciona del modo siguiente:

La aplicación especifica el tamaño de caché de datos de memoria de cada uno de los proveedores que participan en la sesión de sincronización.

Si los dos proveedores especifican un tamaño de caché, Sync Framework utiliza el valor más pequeño para ambos. El tamaño de caché real nunca puede ser superior al 110% del tamaño más pequeño especificado. Durante una sesión de sincronización, si una sola fila supera el 110% del tamaño, la sesión termina con una excepción.

El valor 0 (predeterminado) deshabilita el procesamiento por lotes. Si un proveedor tiene habilitado el procesamiento por lotes y el otro proveedor no, el procesamiento por lotes se habilita para las operaciones de carga y descarga.
La aplicación especifica la ubicación de los archivos en cola de cada proveedor. De forma predeterminada, los archivos en cola se escriben en el directorio temporal de la cuenta bajo la que se ejecuta el proceso de sincronización.
La aplicación llama al método Synchronize.
Sync Framework enumera los cambios de fila en fila. Si se alcanza el tamaño de caché de datos de memoria en el proveedor de origen, los cambios se mantienen en un archivo local en cola y se vacían los datos en memoria. Este proceso se prolonga hasta que se enumeran todos los cambios.
En escenarios con una estructura de n niveles, el código proxy y el servicio de la aplicación transmiten en secuencias los archivos en cola al destino. Para obtener más información, vea la sección Código específico de estructuras de n niveles en este tema. En escenarios con una estructura de dos niveles, el archivo local ya está en el destino, pues, en este caso, todo el código de sincronización se ejecuta en el destino.
Sync Framework deserializa los cambios de los archivos en cola y los aplica. Este proceso se prolonga hasta que todos los cambios se aplican al destino.

Todos los lotes se aplican en una sola transacción. Esa transacción no se crea hasta que el proveedor de destino recibe el último lote.
En escenarios con una estructura de dos niveles, Sync Framework limpia el archivo en cola. En escenarios con una estructura de n niveles, Sync Framework limpia los archivos en cola del equipo en el que se inicia la sincronización, aunque los archivos del nivel intermedio debería limpiarlos el proxy (más adelante en este tema se muestra esta operación en el método Cleanup() de ejemplo). Para administrar los casos en los que se anula una sesión, el nivel intermedio debería utilizar también un proceso para limpiar los archivos que son anteriores a una determinada fecha.

Nota

Los cambios de datos que se aplicarán a un nodo están disponibles mediante la propiedad Context del objeto DbChangesSelectedEventArgs. Cuando los datos no están procesados por lotes, el evento ChangesSelected se genera solamente una vez y todos los cambios están disponibles mediante la propiedad Context. Cuando los datos están procesados por lotes, el evento ChangesSelected se genera para cada lote y solamente los cambios de ese lote están disponibles en ese momento. Si se requieren cambios en todos los lotes, responda a cada evento ChangesSelected y almacene los datos que se devuelven.

En la tabla siguiente se describen los tipos y miembros relacionados con el procesamiento por lotes. La única propiedad requerida para el procesamiento por lotes es MemoryDataCacheSize, pero también se recomienda establecer BatchingDirectory.

Tipo o miembro	Descripción
BatchingDirectory	Obtiene o establece el directorio en el que los archivos por lotes se ponen en cola en el disco. La ruta de acceso especificada debe ser un directorio local del proveedor o proxy que se está ejecutando. No se admiten las rutas de acceso de archivos UNC y las rutas de acceso de URI que no corresponden a archivos. Importante Los archivos en cola contienen los datos de la base de datos sin formato. El directorio en el que se escriben los archivos debe protegerse con los controles de acceso adecuados.
CleanupBatchingDirectory	Obtiene o establece si se deben limpiar los archivos de procesamiento por lotes una vez se han aplicado al destino los cambios de los archivos. El valor predeterminado establece que deben limpiarse los archivos.
MemoryDataCacheSize	Obtiene o establece la cantidad máxima de memoria (en KB) que Sync Framework utiliza para almacenar en la memoria caché los cambios antes de ponerlos en cola en el disco. Nota Esta configuración afecta únicamente al tamaño de los datos y metadatos que se mantienen en memoria para los cambios que se envían al destino. No limita la memoria que utilizan otros componentes de Sync Framework o de la aplicación del usuario.
BatchApplied	Evento que se produce después de que un lote de cambios se ha aplicado en el destino.
BatchSpooled	Evento que se produce después de que cada uno de los lotes de cambios se ha escrito en el disco.
DbBatchAppliedEventArgs	Proporciona los datos para el evento BatchApplied, incluido el número del lote actual y el número total de lotes que se van a aplicar.
DbBatchSpooledEventArgs	Proporciona los datos para el evento BatchSpooled, incluido el número del lote actual y el tamaño del lote.
BatchFileName	Obtiene o establece el nombre del archivo en el que se escriben los cambios en cola.
IsDataBatched	Obtiene o establece si los datos se envían en varios lotes o en un único objeto DataSet.
IsLastBatch	Obtiene o establece si el lote actual es el último lote de cambios.
BatchedDeletesRetried	Obtiene o establece el número de operaciones de eliminación que se reintentaron durante una sesión de sincronización en la que los cambios se procesaron por lotes. Estas operaciones de eliminación vuelven a intentarse en los lotes en el orden en que se produjeron las eliminaciones de la clave principal y la clave externa. Si en el lote actual o en un lote anterior no existe una clave externa, se produce un error al eliminar la clave principal correspondiente. Las operaciones de eliminación en las que se producen errores se reintentan una vez después de aplicarse todos los lotes.
SelectIncrementalChangesCommand (solo es relevante para DbSyncProvider)	Obtiene o establece la consulta o procedimiento almacenado que se utiliza para seleccionar cambios incrementales de la base de datos local. Nota Se recomienda que la consulta que se especifica incluya la cláusula `ORDER BY [sync_row_timestamp]`. Al ordenar las filas en función de su valor de marca de tiempo, se asegura de que, si se reinicia una sesión de sincronización, el proveedor comenzará la enumeración en la marca de tiempo que tiene la marca de agua más elevada (las marcas de agua de cada tabla se preservan con cada lote) y no se pierde ningún cambio.
DataTable	Obtiene o establece el objeto DataTable que contiene los cambios que se van a sincronizar. Si el procesamiento por lotes está habilitado, al obtener acceso a esta propiedad, se deserializa el archivo en cola del disco. Todos los cambios que se realizan en las tablas se preservan en el archivo en cola.
DataSet	Obtiene o establece un objeto DataSet que contiene las filas seleccionadas de la base de datos del mismo nivel. Devuelve NULL si IsDataBatched es true.

Código común para estructuras de dos niveles y n niveles

En los ejemplos de código de esta sección se muestra cómo se administra el procesamiento por lotes en escenarios con estructuras de dos niveles y n niveles. Este código se toma de dos de los ejemplos que se incluyen en el SDK de Sync Framework: SharingAppDemo-CEProviderEndToEnd y WebSharingAppDemo-CEProviderEndToEnd. Cada ejemplo comienza con la ubicación del código, como SharingAppDemo/CESharingForm. Por lo que se refiere al procesamiento por lotes, la diferencia principal entre las dos aplicaciones es el código adicional necesario en el caso de la arquitectura de n niveles para cargar y descargar los archivos en cola y crear directorios para cada nodo que enumera los cambios.

En el ejemplo de código siguiente del controlador de eventos synchronizeBtn_Click de SharingAppDemo/CESharingForm se establece el tamaño de caché de datos de memoria y el directorio en el que deben escribirse los archivos en cola. La ruta de acceso especificada para BatchingDirectory debe estar en un directorio local del proveedor o proxy que se está ejecutando. No se admiten las rutas de acceso de archivos UNC y las rutas de acceso de URI que no corresponden a archivos. La ruta de acceso especificada para BatchingDirectory es el directorio raíz. En cada sesión de sincronización, Sync Framework crea un único subdirectorio en el que se almacenan los archivos en cola de esa sesión. Con el fin de aislar los archivos de diferentes sesiones, este directorio es exclusivo para la combinación origen-destino actual.

En el ejemplo de código siguiente del controlador de eventos synchronizeBtn_Click de WebSharingAppDemo/CESharingForm se establecen las mismas propiedades, pero el directorio de procesamiento por lotes del destino se establece en el proxy y no directamente en el proveedor, como ocurría en el escenario de la estructura de dos niveles:

//Set memory data cache size property. 0 represents non batched mode.
//No need to set memory cache size for Proxy, because the source is 
//enabled for batching: both upload and download will be batched.
srcProvider.MemoryDataCacheSize = this._batchSize;
 

//Set batch spool location. Default value if not set is %Temp% directory.
if (!string.IsNullOrEmpty(this.batchSpoolLocation.Text))
{
    srcProvider.BatchingDirectory = this.batchSpoolLocation.Text;
    destinationProxy.BatchingDirectory = this.batchSpoolLocation.Text;
}

En los ejemplos de código siguientes del archivo SynchronizationHelper de las dos aplicaciones se crean métodos para controlar los eventos BatchSpooled y BatchAppliedEvents que genera un proveedor durante la enumeración y la aplicación de los cambios:

void provider_BatchSpooled(object sender, DbBatchSpooledEventArgs e)
{
    this.progressForm.listSyncProgress.Items.Add("BatchSpooled event fired: Details");
    this.progressForm.listSyncProgress.Items.Add("\tSource Database :" + ((RelationalSyncProvider)sender).Connection.Database);
    this.progressForm.listSyncProgress.Items.Add("\tBatch Name      :" + e.BatchFileName);
    this.progressForm.listSyncProgress.Items.Add("\tBatch Size      :" + e.DataCacheSize);
    this.progressForm.listSyncProgress.Items.Add("\tBatch Number    :" + e.CurrentBatchNumber);
    this.progressForm.listSyncProgress.Items.Add("\tTotal Batches   :" + e.TotalBatchesSpooled);
    this.progressForm.listSyncProgress.Items.Add("\tBatch Watermark :" + ReadTableWatermarks(e.CurrentBatchTableWatermarks));
}

void provider_BatchApplied(object sender, DbBatchAppliedEventArgs e)
{
    this.progressForm.listSyncProgress.Items.Add("BatchApplied event fired: Details");
    this.progressForm.listSyncProgress.Items.Add("\tDestination Database   :" + ((RelationalSyncProvider)sender).Connection.Database);
    this.progressForm.listSyncProgress.Items.Add("\tBatch Number           :" + e.CurrentBatchNumber);
    this.progressForm.listSyncProgress.Items.Add("\tTotal Batches To Apply :" + e.TotalBatchesToApply);
}

//Reads the watermarks for each table from the batch spooled event. //The watermark denotes the max tickcount for each table in each batch.
private string ReadTableWatermarks(Dictionary<string, ulong> dictionary)
{
    StringBuilder builder = new StringBuilder();
    Dictionary<string, ulong> dictionaryClone = new Dictionary<string, ulong>(dictionary);
    foreach (KeyValuePair<string, ulong> kvp in dictionaryClone)
    {
        builder.Append(kvp.Key).Append(":").Append(kvp.Value).Append(",");
    }
    return builder.ToString();
}

Código específico de estructuras de n niveles

Los ejemplos de código restantes se aplican exclusivamente al escenario de la arquitectura de n niveles de WebSharingAppDemo. El código de la arquitectura de n niveles correspondiente está incluido en tres archivos:

El contrato de servicio: IRelationalSyncContract
El servicio web: RelationalWebSyncService
El proxy: RelationalProviderProxy

Los dos proveedores SqlSyncProvider y SqlCeSyncProvider heredan de RelationalSyncProvider, por lo que este código se aplica a ambos. La funcionalidad adicional específica del almacén está separada en los archivos de servicio y proxy de cada tipo de proveedor.

Para entender cómo funciona el procesamiento por lotes en un escenario de n niveles, imagínese una sesión de sincronización en la que el servidor es el origen y el cliente es el destino. Una vez que los cambios se han escrito en el directorio local del servidor, el proceso siguiente tiene lugar en los cambios descargados:

Se llama al método GetChangeBatch en el proxy de cliente. Tal y como se muestra más adelante en el código de ejemplo, este método debería contener código específico para controlar el procesamiento por lotes.
El archivo recibe un archivo por lotes de SqlSyncProvider. El servicio quita toda la información de la ruta de acceso y envía exclusivamente el nombre de archivo a través de la red. De este modo, se evita que la estructura de directorios del servidor quede expuesta a los clientes.
La llamada del proxy a GetChangeBatch devuelve resultados.
1. El proxy detecta que los cambios se procesan por lotes, por lo que llama a DownloadBatchFile mientras pasa el nombre del archivo por lotes como argumento.
2. El proxy crea un único directorio (si no existe ninguno en la sesión) bajo RelationalProviderProxy.BatchingDirectory para retener estos archivos por lotes localmente. El nombre del directorio es el identificador de réplica del elemento del mismo nivel que está enumerando los cambios. De este modo, se asegura de que el proxy y el servicio tienen un único directorio para cada elemento del mismo nivel de la enumeración.
El proxy descarga el archivo y lo almacena localmente. El proxy reemplaza el nombre de archivo en el contexto por la nueva ruta de acceso completa al archivo por lotes del disco local.
El proxy devuelve el contexto al organizador.
Se repiten los pasos 1 a 6 hasta que el proxy recibe el último lote.

El proceso siguiente tiene lugar en los cambios cargados

El organizador llama a ProcessChangeBatch en el proxy.
El proxy determina que se trata de un archivo por lotes, por lo que realiza los pasos siguientes:
1. Quita toda la información de la ruta de acceso y envía exclusivamente el nombre de archivo a través de la red.
2. Llama a HasUploadedBatchFile para determinar si ya se ha cargado el archivo. Si es así, el paso C no es necesario.
3. Si HasUploadedBatchFile devuelve false, llama a UploadBatchFile en el servicio y carga el contenido del archivo por lotes.
  
  El servicio recibirá la llamada a UploadBatchFile y almacenará el lote localmente. La creación del directorio es similar a la del paso 4 anterior.
4. Llama a ApplyChangesen el servicio.
El servidor recibe la llamada de ApplyChanges y determina que se trata de un archivo por lotes. Reemplaza el nombre de archivo en el contexto por la nueva ruta de acceso completa al archivo por lotes del disco local.
El servidor pasa DbSyncContext a SqlSyncProvider local.
Se repiten los pasos 1 a 6 hasta que se envía el último lote.

En el ejemplo de código siguiente de IRelationalSyncContract, se especifican los métodos de carga y descarga que se utilizan para transferir archivos en cola tomando como origen y destino el nivel intermedio:

[OperationContract(IsOneWay = true)]
void UploadBatchFile(string batchFileid, byte[] batchFile);

[OperationContract]
byte[] DownloadBatchFile(string batchFileId);

En los ejemplos de código siguientes de RelationalWebSyncService, se exponen los métodos UploadBatchFile y DownloadBatchFile definidos en el contrato, y se incluye la lógica adicional relativa al procesamiento por lotes de los siguientes métodos:

Cleanup: limpia los archivos en cola de un directorio especificado o del directorio temporal, si no se ha especificado ninguno.
GetChanges: comprueba si los datos están procesados por lotes y, en ese caso, quita la ruta de acceso del directorio del archivo en cola para que no se envíe a través de la red. En escenarios con n niveles, supone un riesgo para la seguridad enviar las rutas de acceso del directorio completas a través de una conexión de red. El nombre de archivo es un GUID.
HasUploadedBatchFile: devuelve si un determinado archivo por lotes ya se ha cargado en el servicio.
ApplyChanges: comprueba si los datos están procesados por lotes, y en ese caso, comprueba si ya se ha cargado el archivo por lotes esperado. Si el archivo no se ha cargado, se produce una excepción. El cliente debería haber cargado el archivo en cola antes de llamar a ApplyChanges.

public abstract class RelationalWebSyncService: IRelationalSyncContract
{
    protected bool isProxyToCompactDatabase;
    protected RelationalSyncProvider peerProvider;
    protected DirectoryInfo sessionBatchingDirectory = null;
    protected Dictionary<string, string> batchIdToFileMapper;
    int batchCount = 0;

    public void Initialize(string scopeName, string hostName)
    {
        this.peerProvider = this.ConfigureProvider(scopeName, hostName);
        this.batchIdToFileMapper = new Dictionary<string, string>();
    }

    public void Cleanup()
    {
        this.peerProvider = null;
        //Delete all file in the temp session directory
        if (sessionBatchingDirectory != null && sessionBatchingDirectory.Exists)
        {
            try
            {
                sessionBatchingDirectory.Delete(true);
            }
            catch 
            { 
                //Ignore 
            }
        }
    }

    public void BeginSession(SyncProviderPosition position)
    {
        Log("*****************************************************************");
        Log("******************** New Sync Session ***************************");
        Log("*****************************************************************");
        Log("BeginSession: ScopeName: {0}, Position: {1}", this.peerProvider.ScopeName, position);
        //Clean the mapper for each session.
        this.batchIdToFileMapper = new Dictionary<string, string>();

        this.peerProvider.BeginSession(position, null/*SyncSessionContext*/);
        this.batchCount = 0;
    }

    public SyncBatchParameters GetKnowledge()
    {
        Log("GetSyncBatchParameters: {0}", this.peerProvider.Connection.ConnectionString);
        SyncBatchParameters destParameters = new SyncBatchParameters();
        this.peerProvider.GetSyncBatchParameters(out destParameters.BatchSize, out destParameters.DestinationKnowledge);
        return destParameters;
    }

    public GetChangesParameters GetChanges(uint batchSize, SyncKnowledge destinationKnowledge)
    {
        Log("GetChangeBatch: {0}", this.peerProvider.Connection.ConnectionString);
        GetChangesParameters changesWrapper = new GetChangesParameters();
        changesWrapper.ChangeBatch  = this.peerProvider.GetChangeBatch(batchSize, destinationKnowledge, out changesWrapper.DataRetriever);

        DbSyncContext context = changesWrapper.DataRetriever as DbSyncContext;
        //Check to see if data is batched
        if (context != null && context.IsDataBatched)
        {
            Log("GetChangeBatch: Data Batched. Current Batch #:{0}", ++this.batchCount);
            //Dont send the file location info. Just send the file name
            string fileName = new FileInfo(context.BatchFileName).Name;
            this.batchIdToFileMapper[fileName] = context.BatchFileName;
            context.BatchFileName = fileName;
        }
        return changesWrapper;
    }

    public SyncSessionStatistics ApplyChanges(ConflictResolutionPolicy resolutionPolicy, ChangeBatch sourceChanges, object changeData)
    {
        Log("ProcessChangeBatch: {0}", this.peerProvider.Connection.ConnectionString);

        DbSyncContext dataRetriever = changeData as DbSyncContext;

        if (dataRetriever != null && dataRetriever.IsDataBatched)
        {
            string remotePeerId = dataRetriever.MadeWithKnowledge.ReplicaId.ToString();
            //Data is batched. The client should have uploaded this file to us prior to calling ApplyChanges.
            //So look for it.
            //The Id would be the DbSyncContext.BatchFileName which is just the batch file name without the complete path
            string localBatchFileName = null;
            if (!this.batchIdToFileMapper.TryGetValue(dataRetriever.BatchFileName, out localBatchFileName))
            {
                //Service has not received this file. Throw exception
                throw new FaultException<WebSyncFaultException>(new WebSyncFaultException("No batch file uploaded for id " + dataRetriever.BatchFileName, null));
            }
            dataRetriever.BatchFileName = localBatchFileName;
        }

        SyncSessionStatistics sessionStatistics = new SyncSessionStatistics();
        this.peerProvider.ProcessChangeBatch(resolutionPolicy, sourceChanges, changeData, new SyncCallbacks(), sessionStatistics);
        return sessionStatistics;
    }

    public void EndSession()
    {
        Log("EndSession: {0}", this.peerProvider.Connection.ConnectionString);
        Log("*****************************************************************");
        Log("******************** End Sync Session ***************************");
        Log("*****************************************************************");
        this.peerProvider.EndSession(null);
        Log("");
    }

    /// <summary>
    /// Used by proxy to see if the batch file has already been uploaded. Optimizes by not resending batch files.
    /// NOTE: This method takes in a file name as an input parameter and hence is suseptible for name canonicalization
    /// attacks. This sample is meant to be a starting point in demonstrating how to transfer sync batch files and is
    /// not intended to be a secure way of doing the same. This SHOULD NOT be used as such in production environment
    /// without doing proper security analysis.
    /// 
    /// Please refer to the following two MSDN whitepapers for more information on guidelines for securing Web servies.
    /// 
    /// Design Guidelines for Secure Web Applications - https://msdn.microsoft.com/en-us/library/aa302420.aspx (Refer InputValidation section)
    /// Architecture and Design Review for Security - https://msdn.microsoft.com/en-us/library/aa302421.aspx (Refer InputValidation section)
    /// </summary>
    /// <param name="batchFileId"></param>
    /// <returns>bool</returns>
    public bool HasUploadedBatchFile(String batchFileId, string remotePeerId)
    {
        this.CheckAndCreateBatchingDirectory(remotePeerId);

        //The batchFileId is the fileName without the path information in it.
        FileInfo fileInfo = new FileInfo(Path.Combine(this.sessionBatchingDirectory.FullName, batchFileId));
        if (fileInfo.Exists && !this.batchIdToFileMapper.ContainsKey(batchFileId))
        {
            //If file exists but is not in the memory id to location mapper then add it to the mapping
            this.batchIdToFileMapper.Add(batchFileId, fileInfo.FullName);
        }
        //Check to see if the proxy has already uploaded this file to the service
        return fileInfo.Exists;
    }

    /// <summary>
    /// NOTE: This method takes in a file name as an input parameter and hence is suseptible for name canonicalization
    /// attacks. This sample is meant to be a starting point in demonstrating how to transfer sync batch files and is
    /// not intended to be a secure way of doing the same. This SHOULD NOT be used as such in production environment
    /// without doing proper security analysis.
    /// 
    /// Please refer to the following two MSDN whitepapers for more information on guidelines for securing Web servies.
    /// 
    /// Design Guidelines for Secure Web Applications - https://msdn.microsoft.com/en-us/library/aa302420.aspx (Refer InputValidation section)
    /// Architecture and Design Review for Security - https://msdn.microsoft.com/en-us/library/aa302421.aspx (Refer InputValidation section)
    /// </summary>
    /// <param name="batchFileId"></param>
    /// <param name="batchContents"></param>
    /// <param name="remotePeerId"></param>
    public void UploadBatchFile(string batchFileId, byte[] batchContents, string remotePeerId)
    {
        Log("UploadBatchFile: {0}", this.peerProvider.Connection.ConnectionString);
        try
        {
            if (HasUploadedBatchFile(batchFileId, remotePeerId))
            {
                //Service has already received this file. So dont save it again.
                return;
            }
            
            //Service hasnt seen the file yet so save it.
            String localFileLocation = Path.Combine(sessionBatchingDirectory.FullName, batchFileId);
            FileStream fs = new FileStream(localFileLocation, FileMode.Create, FileAccess.Write);
            using (fs)
            {
                    fs.Write(batchContents, 0, batchContents.Length);
            }
            //Save this Id to file location mapping in the mapper object
            this.batchIdToFileMapper[batchFileId] = localFileLocation;
        }
        catch (Exception e)
        {
            throw new FaultException<WebSyncFaultException>(new WebSyncFaultException("Unable to save batch file.", e));
        }
    }

    /// <summary>
    /// NOTE: This method takes in a file name as an input parameter and hence is suseptible for name canonicalization
    /// attacks. This sample is meant to be a starting point in demonstrating how to transfer sync batch files and is
    /// not intended to be a secure way of doing the same. This SHOULD NOT be used as such in production environment
    /// without doing proper security analysis.
    /// 
    /// Please refer to the following two MSDN whitepapers for more information on guidelines for securing Web servies.
    /// 
    /// Design Guidelines for Secure Web Applications - https://msdn.microsoft.com/en-us/library/aa302420.aspx (Refer InputValidation section)
    /// Architecture and Design Review for Security - https://msdn.microsoft.com/en-us/library/aa302421.aspx (Refer InputValidation section)
    /// </summary>
    /// <param name="batchFileId"></param>
    /// <returns></returns>
    public byte[] DownloadBatchFile(string batchFileId)
    {
        try
        {
            Log("DownloadBatchFile: {0}", this.peerProvider.Connection.ConnectionString);
            Stream localFileStream = null;

            string localBatchFileName = null;

            if (!this.batchIdToFileMapper.TryGetValue(batchFileId, out localBatchFileName))
            {
                throw new FaultException<WebSyncFaultException>(new WebSyncFaultException("Unable to retrieve batch file for id." + batchFileId, null));
            }

            localFileStream = new FileStream(localBatchFileName, FileMode.Open, FileAccess.Read);
            byte[] contents = new byte[localFileStream.Length];
            localFileStream.Read(contents, 0, contents.Length);
            return contents;
        }
        catch (Exception e)
        {
            throw new FaultException<WebSyncFaultException>(new WebSyncFaultException("Unable to read batch file for id " + batchFileId, e));
        }
    }

    protected void Log(string p, params object[] paramArgs)
    {
        Console.WriteLine(p, paramArgs);
    }

    //Utility functions that the sub classes need to implement.
    protected abstract RelationalSyncProvider ConfigureProvider(string scopeName, string hostName);


    private void CheckAndCreateBatchingDirectory(string remotePeerId)
    {
        //Check to see if we have temp directory for this session.
        if (sessionBatchingDirectory == null)
        {
            //Generate a unique Id for the directory
            //We use the peer id of the store enumerating the changes so that the local temp directory is same for a given source
            //across sync sessions. This enables us to restart a failed sync by not downloading already received files.
            string sessionDir = Path.Combine(this.peerProvider.BatchingDirectory, "WebSync_" + remotePeerId);
            sessionBatchingDirectory = new DirectoryInfo(sessionDir);
            //Create the directory if it doesnt exist.
            if (!sessionBatchingDirectory.Exists)
            {
                sessionBatchingDirectory.Create();
            }
        }
    }
}

En los ejemplos de código siguientes de RelationalProviderProxy, se establecen las propiedades y se llama a los métodos del servicio web:

BatchingDirectory: permite a la aplicación establecer el directorio de procesamiento por lotes para el nivel intermedio.
EndSession: limpia los archivos en cola de un directorio especificado.
GetChangeBatch: descarga los lotes de cambios llamando al método DownloadBatchFile.
ProcessChangeBatch: carga los lotes de cambios llamando al método UploadBatchFile.

public abstract class RelationalProviderProxy : KnowledgeSyncProvider, IDisposable
{
    protected IRelationalSyncContract proxy;
    protected SyncIdFormatGroup idFormatGroup;
    protected string scopeName;
    protected DirectoryInfo localBatchingDirectory;

    //Represents either the SQL server host name or the CE database file name. Sql database name
    //is always peer1
    //For this sample scopeName is always Sales
    protected string hostName;

    private string batchingDirectory = Environment.ExpandEnvironmentVariables("%TEMP%");

    public string BatchingDirectory
    {
        get { return batchingDirectory; }
        set 
        {
            if (string.IsNullOrEmpty(value))
            {
                throw new ArgumentException("value cannot be null or empty");
            }
            try
            {
                Uri uri = new Uri(value);
                if (!uri.IsFile || uri.IsUnc)
                {
                    throw new ArgumentException("value must be a local directory");
                }
                batchingDirectory = value;
            }
            catch (Exception e)
            {
                throw new ArgumentException("Invalid batching directory.", e);
            }
        }
    }

    public RelationalProviderProxy(string scopeName, string hostName)
    {
        this.scopeName = scopeName;
        this.hostName = hostName;
        this.CreateProxy();            
        this.proxy.Initialize(scopeName, hostName);
    }

    public override void BeginSession(SyncProviderPosition position, SyncSessionContext syncSessionContext)
    {
        this.proxy.BeginSession(position);
    }

    public override void EndSession(SyncSessionContext syncSessionContext)
    {
        proxy.EndSession();
        if (this.localBatchingDirectory != null && this.localBatchingDirectory.Exists)
        {
            //Cleanup batching releated files from this session
            this.localBatchingDirectory.Delete(true);
        }
    }

    public override ChangeBatch GetChangeBatch(uint batchSize, SyncKnowledge destinationKnowledge, out object changeDataRetriever)
    {
        GetChangesParameters changesWrapper = proxy.GetChanges(batchSize, destinationKnowledge);
        //Retrieve the ChangeDataRetriever and the ChangeBatch
        changeDataRetriever = changesWrapper.DataRetriever;

        DbSyncContext context = changeDataRetriever as DbSyncContext;
        //Check to see if the data is batched.
        if (context != null && context.IsDataBatched)
        {
            if (this.localBatchingDirectory == null)
            {
                //Retrieve the remote peer id from the MadeWithKnowledge.ReplicaId. MadeWithKnowledge is the local knowledge of the peer 
                //that is enumerating the changes.
                string remotePeerId = context.MadeWithKnowledge.ReplicaId.ToString();

                //Generate a unique Id for the directory.
                //We use the peer id of the store enumerating the changes so that the local temp directory is same for a given source
                //across sync sessions. This enables us to restart a failed sync by not downloading already received files.
                string sessionDir = Path.Combine(this.batchingDirectory, "WebSync_" + remotePeerId);
                this.localBatchingDirectory = new DirectoryInfo(sessionDir);
                //Create the directory if it doesnt exist.
                if (!this.localBatchingDirectory.Exists)
                {
                    this.localBatchingDirectory.Create();
                }
            }

            string localFileName = Path.Combine(this.localBatchingDirectory.FullName, context.BatchFileName);
            FileInfo localFileInfo = new FileInfo(localFileName);
            
            //Download the file only if doesnt exist
            FileStream localFileStream = new FileStream(localFileName, FileMode.Create, FileAccess.Write);
            if (!localFileInfo.Exists)
            {
                byte[] remoteFileContents = this.proxy.DownloadBatchFile(context.BatchFileName);
                using (localFileStream)
                {
                    localFileStream.Write(remoteFileContents, 0, remoteFileContents.Length);
                }
            }
            //Set DbSyncContext.Batchfile name to the new local file name
            context.BatchFileName = localFileName;
        }

        return changesWrapper.ChangeBatch;
    }

    public override FullEnumerationChangeBatch GetFullEnumerationChangeBatch(uint batchSize, SyncId lowerEnumerationBound, SyncKnowledge knowledgeForDataRetrieval, out object changeDataRetriever)
    {
        throw new NotImplementedException();
    }

    public override void GetSyncBatchParameters(out uint batchSize, out SyncKnowledge knowledge)
    {
        SyncBatchParameters wrapper = proxy.GetKnowledge();
        batchSize = wrapper.BatchSize;
        knowledge = wrapper.DestinationKnowledge;
    }

    public override SyncIdFormatGroup IdFormats
    {
        get
        {
            if (idFormatGroup == null)
            {
                idFormatGroup = new SyncIdFormatGroup();

                //
                // 1 byte change unit id (Harmonica default before flexible ids)
                //
                idFormatGroup.ChangeUnitIdFormat.IsVariableLength = false;
                idFormatGroup.ChangeUnitIdFormat.Length = 1;

                //
                // Guid replica id
                //
                idFormatGroup.ReplicaIdFormat.IsVariableLength = false;
                idFormatGroup.ReplicaIdFormat.Length = 16;


                //
                // Sync global id for item ids
                //
                idFormatGroup.ItemIdFormat.IsVariableLength = true;
                idFormatGroup.ItemIdFormat.Length = 10 * 1024;
            }

            return idFormatGroup;
        }
    }

    public override void ProcessChangeBatch(ConflictResolutionPolicy resolutionPolicy, ChangeBatch sourceChanges, object changeDataRetriever, SyncCallbacks syncCallbacks, SyncSessionStatistics sessionStatistics)
    {
        DbSyncContext context = changeDataRetriever as DbSyncContext;
        if (context != null && context.IsDataBatched)
        {
            string fileName = new FileInfo(context.BatchFileName).Name;

            //Retrieve the remote peer id from the MadeWithKnowledge.ReplicaId. MadeWithKnowledge is the local knowledge of the peer 
            //that is enumerating the changes.
            string peerId = context.MadeWithKnowledge.ReplicaId.ToString();

            //Check to see if service already has this file
            if (!this.proxy.HasUploadedBatchFile(fileName, peerId))
            {
                //Upload this file to remote service
                FileStream stream = new FileStream(context.BatchFileName, FileMode.Open, FileAccess.Read);
                byte[] contents = new byte[stream.Length];
                using (stream)
                {
                    stream.Read(contents, 0, contents.Length);
                }
                this.proxy.UploadBatchFile(fileName, contents, peerId);
            }

            context.BatchFileName = fileName;
        }
        this.proxy.ApplyChanges(resolutionPolicy, sourceChanges, changeDataRetriever);
    }

    public override void ProcessFullEnumerationChangeBatch(ConflictResolutionPolicy resolutionPolicy, FullEnumerationChangeBatch sourceChanges, object changeDataRetriever, SyncCallbacks syncCallbacks, SyncSessionStatistics sessionStatistics)
    {
        throw new NotImplementedException();
    }

    protected abstract void CreateProxy();

    #region IDisposable Members

    public void Dispose()
    {
        this.proxy.Cleanup();
        this.proxy = null;
        GC.SuppressFinalize(this);
    }

    #endregion
}

Vea también

Conceptos

Sincronizar SQL Server y SQL Server Compact

Compartir a través de