Biztalk Acid Properties

The atomic scope was designed to handle Atomicity, Consistency,
Isolation, and Durability (ACID) compliant operations that must either all succeed or
all fail as a group. This is a classic database transaction style. It is designed to carry
an orchestration from one stable state to another.

Atomicity

A transaction represents an atomic unit of work. Either all modifications within a transaction are performed, or none of the modifications are performed.

Consistency

When committed, a transaction must preserve the integrity of the data within the system. If a transaction performs a data modification on a database that was internally consistent before the transaction started, the database must still be internally consistent when the transaction is committed. Ensuring this property is largely the responsibility of the application developer.

Isolation

Modifications made by concurrent transactions must be isolated from the modifications made by other concurrent transactions. Isolated transactions that run concurrently perform modifications that preserve internal database consistency exactly as they would if the transactions were run serially.

Durability

After a transaction is committed, all modifications are permanently in place in the system by default. The modifications persist even if a system failure occurs.

If you absolutely must call a non-Serializable class, and can only do

it in an atomic scope, try to combine this with other operations to make the most of

the trip to the message box; like a send shape as shown in the following figure:

XLANG Message.

If you need to pass non-XML messages through an orchestration, you should really use XLANGMessage to do this.

Represents a message instance that is declared with an XLANG service.

All public members of this type are safe for multithreaded operations.

Better to read xml body part by using XMLReader rather than as XMLDocument.


public XmlReader func(XLANGMessage message)
{
message[0].RetrieveAs(typeof(XmlReader)) as XmlReader;
}

BIZTALK Canonical Schema

Every external schema should have a corresponding internal schema or translate to part of a composite internal schema. These schemas are commonly referred to as canoncial schema.

EDI POST PROCESS PIPELINE - 4 (SeekableReadOnlyStream)

//---------------------------------------------------------------------
// File: SeekableReadOnlyStream.cs
//
// Summary: A sample pipeline component which demonstrates how to promote message context
//          properties and write distinguished fields for XML messages using arbitrary
//          XPath expressions.
//
// Sample: Arbitrary XPath Property Handler Pipeline Component SDK
//
//---------------------------------------------------------------------
// This file is part of the Microsoft BizTalk Server 2009 SDK
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// This source code is intended only as a supplement to Microsoft BizTalk
// Server 2009 release and/or on-line documentation. See these other
// materials for detailed information regarding Microsoft code samples.
//
// THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY
// KIND, WHETHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR
// PURPOSE.
//---------------------------------------------------------------------

using System;
using System.IO;
using System.Diagnostics;

namespace Microsoft.Samples.BizTalk.Pipelines.CustomComponent
{
/// <summary>
/// Implements a seekable read-only stream which uses buffering if
/// underlying stream is not seekable. Buffer in memory has size
/// threshold and overflows to disk (temporary file) if number of bytes.
/// </summary>
public class SeekableReadOnlyStream : Stream
{
/// <summary>
/// Initializes a SeekableReadOnlyStream instance with base stream and
/// buffering stream.
/// </summary>
/// <param name="baseStream">Base stream</param>
/// <param name="overflowStream">Buffering stream</param>
public SeekableReadOnlyStream(Stream baseStream, Stream bufferingStream)
{
if (null == baseStream)
throw new ArgumentNullException("baseStream");
if (null == bufferingStream)
throw new ArgumentNullException("bufferingStream");

// Sanity check - make sure that buffering stream is seekable
if (!bufferingStream.CanSeek)
throw new NotSupportedException("Buffering stream must be seekable");

this.baseStream = baseStream;
this.bufferingStream = bufferingStream;
}

/// <summary>
/// Initializes a SeekableReadOnlyStream instance with base stream and inherently uses
/// VirtualStream instance as buffering stream.
/// </summary>
/// <param name="baseStream">Base stream</param>
public SeekableReadOnlyStream(Stream baseStream) : this(baseStream, new VirtualStream())
{
// Empty
}

/// <summary>
/// Initializes a SeekableReadOnlyStream instance with base stream and buffer size, and
/// inherently uses VirtualStream instance as buffering stream.
/// </summary>
/// <param name="baseStream">Base stream</param>
/// <param name="bufferSize">Buffer size</param>
public SeekableReadOnlyStream(Stream baseStream, int bufferSize) : this(baseStream, new VirtualStream(bufferSize))
{
// Empty
}

/// <summary>
/// Gets a flag indicating whether this stream can be read.
/// </summary>
public override bool CanRead
{
get { return true; }
}

/// <summary>
/// Gets a flag indicating whether this stream can be written to.
/// </summary>
public override bool CanWrite
{
get { return false; }
}

public override bool CanSeek
{
get { return true; }
}

/// <summary>
/// Gets or sets a stream position.
/// </summary>
public override long Position
{
get
{
// Check if base stream is seekable
if (baseStream.CanSeek)
return baseStream.Position;

return bufferingStream.Position;
}
set
{
// Check if base stream is seekable
if (baseStream.CanSeek)
{
baseStream.Position = value;
return;
}

// Check if current position is the same as being set
if (bufferingStream.Position == value)
return;

// Check if stream position is being set to the value which is in already
// read to the buffering stream space, i.e. less than current buffering stream
// position or less than length of the buffering stream
if (value < bufferingStream.Position || value < bufferingStream.Length)
{
// Just change position in the buffering stream
bufferingStream.Position = value;
}
else
{
//
// Need to read buffer from the base stream from the current position in
// base stream to the position being set and write that buffer to the end
// of the buffering stream
//

// Set position to the last byte in the buffering stream
bufferingStream.Seek(0, SeekOrigin.End);

// Read buffer from the base stream and write it to the buffering stream
// in 4K chunks
byte [] buffer = new byte[ 4096 ];
long bytesToRead = value - bufferingStream.Position;
while (bytesToRead > 0)
{
// Read to buffer 4K or byteToRead, whichever is less
int bytesRead = baseStream.Read(buffer, 0, (int) Math.Min(bytesToRead, buffer.Length));

// Check if any bytes were read
if (0 == bytesRead)
break;

// Write read bytes to the buffering stream
bufferingStream.Write(buffer, 0, bytesRead);

// Decrease bytes to read counter
bytesToRead -= bytesRead;
}

//
// Since this stream is not writable, any attempt to point Position beyond the length
// of the base stream will not succeed, and buffering stream position will be set to the
// last byte in the buffering stream.
//
}
}
}

/// <summary>
/// Seeks in stream. For this stream can be very expensive because entire base stream
/// can be dumped into buffering stream if SeekOrigin.End is used.
/// </summary>
/// <param name="offset">A byte offset relative to the origin parameter</param>
/// <param name="origin">A value of type SeekOrigin indicating the reference point used to obtain the new position</param>
/// <returns>The new position within the current stream</returns>
public override long Seek(long offset, SeekOrigin origin)
{
// Check if base stream is seekable
if (baseStream.CanSeek)
return baseStream.Seek(offset, origin);

if (SeekOrigin.Begin == origin)
{
// Just set the absolute position using Position property
Position = offset;
return Position;
}

if (SeekOrigin.Current == origin)
{
// Set the position using current Position property value plus offset
Position = Position + offset;
return Position;
}

if (SeekOrigin.End == origin)
{
//
// Need to read all remaining not read bytes from the base stream to the
// buffering stream. We can't use offset here because stream size may not
// be available because it's not seekable. Then we'll set the position
// based on buffering stream size.
//

// Set position to the last byte in the buffering stream
bufferingStream.Seek(0, SeekOrigin.End);

// Read all remaining bytes from the base stream to the buffering stream
byte [] buffer = new byte[ 4096 ];
for (;;)
{
// Read buffer from base stream
int bytesRead = baseStream.Read(buffer, 0, buffer.Length);

// Break the reading loop if the base stream is exhausted
if (0 == bytesRead)
break;

// Write buffer to the buffering stream
bufferingStream.Write(buffer, 0, bytesRead);
}

                // Now buffering stream size is equal to the base stream size. Set position
// using begin origin
Position = bufferingStream.Length - offset;
return Position;
}

throw new NotSupportedException("Not supported SeekOrigin");
}

/// <summary>
/// Gets the length in bytes of the stream. For this stream can be very expensive
/// because entire base stream will be dumped into buffering stream.
/// </summary>
public override long Length
{
get
{
// Check if base stream is seekable
if (baseStream.CanSeek)
return baseStream.Length;

// Preserve the current stream position
long position = Position;

// Seek to the end of stream
Seek(0, SeekOrigin.End);

// Length will be equal to the current position
long length = Position;

// Restore the current stream position
Position = position;

return length;
}
}

/// <summary>
/// Reads a sequence of bytes from the current stream and advances the position within the stream by the number of bytes read.
/// </summary>
/// <param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between offset and (offset + count- 1) replaced by the bytes read from the current source</param>
/// <param name="offset">The zero-based byte offset in buffer at which to begin storing the data read from the current stream</param>
/// <param name="count">The maximum number of bytes to be read from the current stream</param>
/// <returns>The total number of bytes read into the buffer. This can be less than the number of bytes requested if that many bytes are not currently available, or zero (0) if the end of the stream has been reached</returns>
public override int Read(byte[] buffer, int offset, int count)
{
// Check if base stream is seekable
if (baseStream.CanSeek)
return baseStream.Read(buffer, offset, count);

int bytesReadTotal = 0;

// Check if buffering stream has some bytes to read, starting from the
// current position
if (bufferingStream.Length > bufferingStream.Position)
{
// Read available bytes in buffering stream or count bytes to the buffer, whichever is less
bytesReadTotal = bufferingStream.Read(buffer, offset, (int) Math.Min(bufferingStream.Length - bufferingStream.Position, count));

// Account for bytes read from the buffering stream
count -= bytesReadTotal;
offset += bytesReadTotal;
}

// Check if we have any more bytes to read
if (count > 0)
{
Debug.Assert(bufferingStream.Position == bufferingStream.Length);

//
// At this point, buffering stream has position set to its end. We need to read buffer from
// the base stream and write it to the buffering stream
//

// Read count bytes from the base stream starting from offset
int bytesRead = baseStream.Read(buffer, offset, count);

// Check if bytes were really read
if (bytesRead > 0)
{
// Write number of read bytes to the buffering stream starting from offset in buffer
bufferingStream.Write(buffer, offset, bytesRead);
}

// Add number of bytes read at this step to the number of totally read bytes
bytesReadTotal += bytesRead;
}

return bytesReadTotal;
}

/// <summary>
/// Writes to stream.
/// </summary>
/// <param name="buffer">Buffer to write to stream</param>
/// <param name="offset">Stream offset to start write from</param>
/// <param name="count">Number of bytes from buffer to write</param>
/// <exception cref="NotSupportedException">Is thrown always</exception>
public override void Write(byte[] buffer, int offset, int count)
{
throw new NotSupportedException();
}

/// <summary>
/// Set stream length.
/// </summary>
/// <param name="value">Stream length</param>
/// <exception cref="NotSupportedException">Is thrown always</exception>
public override void SetLength(long value)
{
throw new NotSupportedException();
}

/// <summary>
/// Closes base and buffering streams.
/// </summary>
public override void Close()
{
// Close underlying streams
baseStream.Close();
bufferingStream.Close();
}

/// <summary>
/// Flushes the stream.
/// </summary>
public override void Flush()
{
// Flush the buffering stream
bufferingStream.Flush();
}


private Stream baseStream;
private Stream bufferingStream;
}
}

EDI POST PROCESS PIPELINE - 3 VirtualStream

//---------------------------------------------------------------------
// File: VirtualStream.cs
//
// Summary: A sample pipeline component which demonstrates how to promote message context
//          properties and write distinguished fields for XML messages using arbitrary
//          XPath expressions.
//
// Sample: Arbitrary XPath Property Handler Pipeline Component SDK
//
//---------------------------------------------------------------------
// This file is part of the Microsoft BizTalk Server 2009 SDK
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// This source code is intended only as a supplement to Microsoft BizTalk
// Server 2009 release and/or on-line documentation. See these other
// materials for detailed information regarding Microsoft code samples.
//
// THIS CODE AND INFORMATION ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY
// KIND, WHETHER EXPRESSED OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND/OR FITNESS FOR A PARTICULAR
// PURPOSE.
//---------------------------------------------------------------------

using System;
using System.IO;
using System.Text;
using System.Runtime.InteropServices;
using Microsoft.Win32.SafeHandles;

namespace Microsoft.Samples.BizTalk.Pipelines.CustomComponent
{
/// <summary>
/// Implements a virtual stream, i.e. the always seekable stream which
/// uses configurable amount of memory to reduce a memory footprint and
/// temporarily stores remaining data in a temporary file on disk.
/// </summary>
public sealed class VirtualStream : Stream, IDisposable
{
/// <summary>
/// Memory handling.
/// </summary>
public enum MemoryFlag
{
AutoOverFlowToDisk = 0,
OnlyInMemory = 1,
OnlyToDisk = 2
}

// Constants
private const int MemoryThreshold = 4*1024*1024; // The maximum possible memory consumption (4Mb)
private const int DefaultMemorySize = 4*1024; // Default memory consumption (4Kb)

private Stream wrappedStream;
private bool isDisposed;
private bool isInMemory;
private int thresholdSize;
private MemoryFlag memoryStatus;

/// <summary>
/// Initializes a VirtualStream instance with default parameters (10K memory buffer,
/// allow overflow to disk).
/// </summary>
public VirtualStream()
: this(DefaultMemorySize, MemoryFlag.AutoOverFlowToDisk, new MemoryStream())
{
}

/// <summary>
/// Initializes a VirtualStream instance with memory buffer size.
/// </summary>
/// <param name="bufferSize">Memory buffer size</param>
public VirtualStream(int bufferSize)
: this(bufferSize, MemoryFlag.AutoOverFlowToDisk, new MemoryStream(bufferSize))
{
}

/// <summary>
/// Initializes a VirtualStream instance with a default memory size and memory flag specified.
/// </summary>
/// <param name="flag">Memory flag</param>
public VirtualStream(MemoryFlag flag)
: this(DefaultMemorySize, flag,
(flag == MemoryFlag.OnlyToDisk) ? CreatePersistentStream() : new MemoryStream())
{
}

/// <summary>
/// Initializes a VirtualStream instance with a memory buffer size and memory flag specified.
/// </summary>
/// <param name="bufferSize">Memory buffer size</param>
/// <param name="flag">Memory flag</param>
public VirtualStream(int bufferSize, MemoryFlag flag)
: this(bufferSize, flag,
(flag == MemoryFlag.OnlyToDisk) ? CreatePersistentStream() : new MemoryStream(bufferSize))
{
}

/// <summary>
/// Initializes a VirtualStream instance with a memory buffer size, memory flag and underlying stream
/// specified.
/// </summary>
/// <param name="bufferSize">Memory buffer size</param>
/// <param name="flag">Memory flag</param>
/// <param name="dataStream">Underlying stream</param>
private VirtualStream(int bufferSize, MemoryFlag flag, Stream dataStream)
{
if (null == dataStream)
throw new ArgumentNullException("dataStream");

isInMemory = (flag != MemoryFlag.OnlyToDisk);
memoryStatus = flag;
bufferSize = Math.Min(bufferSize, MemoryThreshold);
thresholdSize = bufferSize;

if (isInMemory)
wrappedStream = dataStream;  // Don't want to double wrap memory stream
else
wrappedStream = new BufferedStream(dataStream, bufferSize);
isDisposed = false;
}

#region Stream Methods and Properties

/// <summary>
/// Gets a flag indicating whether a stream can be read.
/// </summary>
override public bool CanRead
{
get {return wrappedStream.CanRead;}
}
/// <summary>
/// Gets a flag indicating whether a stream can be written.
/// </summary>
override public bool CanWrite
{
get {return wrappedStream.CanWrite;}
}
/// <summary>
/// Gets a flag indicating whether a stream can seek.
/// </summary>
override public bool CanSeek
{
get {return true;}
}
/// <summary>
/// Returns the length of the source stream.
/// <seealso cref="GetLength()"/>
/// </summary>
override public long Length
{
get {return wrappedStream.Length;}
}

/// <summary>
/// Gets or sets a position in the stream.
/// </summary>
override public long Position
{
get {return wrappedStream.Position;}
set {wrappedStream.Seek(value, SeekOrigin.Begin);}
}

/// <summary>
/// <see cref="Stream.Close()"/>
/// </summary>
/// <remarks>
/// Calling other methods after calling Close() may result in a ObjectDisposedException beeing throwed.
/// </remarks>
override public void Close()
{
if(!isDisposed)
{
GC.SuppressFinalize(this);
Cleanup();
}
}

/// <summary>
/// <see cref="Stream.Flush()"/>
/// </summary>
/// <remarks>
/// </remarks>
override public void Flush()
{
ThrowIfDisposed();
wrappedStream.Flush();
}

/// <summary>
/// <see cref="Stream.Read()"/>
/// </summary>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="count"></param>
/// <returns>
/// The number of bytes read
/// </returns>
/// <remarks>
/// May throw <see cref="ObjectDisposedException"/>.
/// It will read from cached persistence stream
/// </remarks>
override public int Read(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();
return wrappedStream.Read(buffer, offset, count);
}

/// <summary>
/// <see cref="Stream.Seek()"/>
/// </summary>
/// <param name="offset"></param>
/// <param name="origin"></param>
/// <returns>
/// The current position
/// </returns>
/// <remarks>
/// May throw <see cref="ObjectDisposedException"/>.
/// It will cache any new data into the persistence stream
/// </remarks>
override public long Seek(long offset, SeekOrigin origin)
{
ThrowIfDisposed();
return wrappedStream.Seek(offset, origin);
}

/// <summary>
/// <see cref="Stream.SetLength()"/>
/// </summary>
/// <param name="length"></param>
/// <remarks>
/// May throw <see cref="ObjectDisposedException"/>.
/// </remarks>
override public void SetLength(long length)
{
ThrowIfDisposed();

// Check if new position is greater than allowed by threshold
if (memoryStatus == MemoryFlag.AutoOverFlowToDisk &&
isInMemory &&
length > thresholdSize)
{
// Currently in memory, and the new write will push it over the limit
// Switching to Persist Stream
BufferedStream persistStream = new BufferedStream(CreatePersistentStream(), thresholdSize);

// Copy current wrapped memory stream to the persist stream
CopyStreamContent((MemoryStream)wrappedStream, persistStream);

// Close old wrapped stream
if (wrappedStream != null)
wrappedStream.Close();

wrappedStream = persistStream;
isInMemory = false;
}

// Set new length for the wrapped stream
wrappedStream.SetLength(length);
}

/// <summary>
/// <see cref="Stream.Write()"/>
/// <param name="buffer"></param>
/// <param name="offset"></param>
/// <param name="count"></param>
/// </summary>
/// <remarks>
/// Write to the underlying stream.
/// </remarks>
override public void Write(byte[] buffer, int offset, int count)
{
ThrowIfDisposed();

// Check if new position after write is greater than allowed by threshold
if (memoryStatus == MemoryFlag.AutoOverFlowToDisk &&
isInMemory &&
(count + wrappedStream.Position) > thresholdSize)
{
// Currently in memory, and the new write will push it over the limit
// Switching to Persist Stream
BufferedStream persistStream = new BufferedStream(CreatePersistentStream(), thresholdSize);

// Copy current wrapped memory stream to the persist stream
CopyStreamContent((MemoryStream) wrappedStream, persistStream);

// Close old wrapped stream
if (wrappedStream != null)
wrappedStream.Close();

wrappedStream = persistStream;
isInMemory = false;
}

wrappedStream.Write(buffer, offset, count);
}

#endregion

#region IDisposable Interface

/// <summary>
/// <see cref="IDisposeable.Dispose()"/>
/// </summary>
/// <remarks>
/// It will call <see cref="Close()"/>
/// </remarks>
public void Dispose()
{
Close();
}

#endregion

#region Private Utility Functions

/// <summary>
/// Utility method called by the Finalize(), Close() or Dispose() to close and release
/// both the source and the persistence stream.
/// </summary>
private void Cleanup()
{
if(!isDisposed)
{
isDisposed = true;
if(null != wrappedStream)
{
wrappedStream.Close();
wrappedStream = null;
}
}
}

/// <summary>
/// Copies source memory stream to the target stream.
/// </summary>
/// <param name="source">Source memory stream</param>
/// <param name="target">Target stream</param>
private void CopyStreamContent(MemoryStream source, Stream target)
{
// Remember position for the source stream
long currentPosition = source.Position;

// Read and write in chunks each thresholdSize
byte[] tempBuffer = new Byte[thresholdSize];
int read = 0;

source.Position = 0;
while ((read = source.Read(tempBuffer, 0, tempBuffer.Length)) != 0)
target.Write(tempBuffer, 0, read);

// Set target's stream position to be the same as was in source stream. This is required because
// target stream is going substitute source stream.
target.Position = currentPosition;

// Restore source stream's position (just in case to preserve the source stream's state)
source.Position = currentPosition;
}

/// <summary>
/// Called by other methods to check the stream state.
/// It will thorw <see cref="ObjectDisposedException"/> if the stream was closed or disposed.
/// </summary>
private void ThrowIfDisposed()
{
if(isDisposed || null == wrappedStream)
throw new ObjectDisposedException("VirtualStream");
}

/// <summary>
/// Utility method.
/// Creates a FileStream with a unique name and the temporary and delete-on-close attributes.
/// </summary>
/// <returns>
/// The temporary persistence stream
/// </returns>
public static Stream CreatePersistentStream()
{
StringBuilder name = new StringBuilder(256);

IntPtr handle;
if(0 == GetTempFileName(Path.GetTempPath(), "BTS", 0, name))
throw new IOException("GetTempFileName Failed.", Marshal.GetHRForLastWin32Error());

handle = CreateFile(name.ToString(), (UInt32) FileAccess.ReadWrite, 0, IntPtr.Zero, (UInt32) FileMode.Create, 0x04000100, IntPtr.Zero);

// FileStream constructor will throw exception if handle is zero or -1.
return new FileStream(new SafeFileHandle(handle, true), FileAccess.ReadWrite);
}

[DllImport("kernel32.dll")]
private static extern UInt32 GetTempFileName
(
string path,
string prefix,
UInt32 unique,
StringBuilder name
);

[DllImport("kernel32.dll")]
private static extern IntPtr CreateFile
(
string name,
UInt32 accessMode,
UInt32 shareMode,
IntPtr security,
UInt32 createMode,
UInt32 flags,
IntPtr template
);

#endregion
}
}

Isolated Vs In Process Host

In Process host instances run the Biztalk run time executable, which covers everything we discussed earlier about host instances.  It is a Windows service running on the specific server with in the Biztlak group.  An isolated host instance is isolated in the sense that it runs in another process context, most often IIS. This is isolated, and called such, because it has a limited role that it can fulfill as it is not the BizTalk service executable we described previously.  It is generally a host specifically for receive adapter and cannot perform orchestration,tracking or send operations because isolated host instances are not in BizTalk run-time.

Biztalk Group

This BizTalk group is the top level abstraction of BizTalk.  It is a logical container for everything in a Bizatalk installation. The group generally has an analogy to an enterprise,  department , or division of the organization that it serves.