Mindmajix

Developing the Destination Adapter

Building the Destination Adapter

The requirement for the Destination adapter is that it accepts an input from an upstream component of any description and converts it to a format similar to that seen in the Source adapter. The component will use a FILE Connection Manager, and as shown in earlier components, this involves a significant amount of validation. You also need to validate whether the component is structurally correct; if it isn’t, you need to correct things.

If you’re following along with the tutorial and are writing the code manually, right-click on the Wrox. Pipeline project in Visual Studio and click “Add ⇒ New Item”. Select the “Class” template in the “Add New Item” dialog and create a new file named ProfSSISDestAdapter.cs.

The first thing you always need to do is declare some variables that will be used throughout the component (ProfSSISDestAdapter.cs). You also need to create the very valuable state-information struct that is going to store the details of the columns, which will be needed in PreExecute and ProcessInput 

#region Variables
private ArrayList _columnInfos = new ArrayList();
private Microsoft.SqlServer.Dts.Runtime.DTSFileConnectionUsageType
_fil;
private string _filename;
FileStream _fs;
StreamWriter _sw;
#endregion

You should quickly run through the meaning of these variables and when they will be needed. The _columnInfos variable is used to store ColumnInfo objects, which describe the columns in the InputColumnCollection. The _fil variable is used to validate the type of FILE Connection Manager the user has assigned to your component. _filename stores the name of the file that is retrieved from the FILE Connection Manager. The final two variables, _fs and _sw, are used when you write to the text file in ProcessInput. Now take a look at the ColumnInfo struct:

#region ColumnInfo
private struct ColumnInfo
{
Public int BufferColumnIndex;
public string ColumnName;
}
#endregion

The struct is used to store the index number of the column in the buffer and the name of the column.

At this point, it is time to look at the ProvideComponentProperties method, which is where you set up the component and prepare it for use by an SSIS package, as in the other two components. Here’s the method in full (ProfSSISDestAdapter.cs):

public override void ProvideComponentProperties()
{
ComponentMetaData.RuntimeConnectionCollection.RemoveAll();
RemoveAllInputsOutputsAndCustomProperties();
ComponentMetaData.Name = “Wrox SSIS Destination Adapter”;
ComponentMetaData.Description = “Our first Destination Adapter”;
ComponentMetaData.ContactInfo = “www.wrox.com”;
IDTSRuntimeConnection100 rtc =
ComponentMetaData.RuntimeConnectionCollection.New();
rtc.Name = “File To Write”;
rtc.Description = “This is the file to which we want to write”;
IDTSInput100 input = ComponentMetaData.InputCollection.New();
input.Name = “Component Input”;
input.Description = “This is what we see from the upstream
component”;
input.HasSideEffects = true;
}

The first part of the method gets rid of any runtime Connection Managers that the component may have and removes any custom properties, inputs, and outputs it may contain. This makes the component a clean slate to which you can now add back anything it may need:

ComponentMetaData.RuntimeConnectionCollection.RemoveAll();
RemoveAllInputsOutputsAndCustomProperties();

The component requires one connection, defined as follows:

IDTSRuntimeConnection100 rtc =
ComponentMetaData.RuntimeConnectionCollection.New();
rtc.Name = “File To Write”;
rtc.Description = “This is the file to which we want to write”;

The preceding piece of code gives the user the opportunity to specify a Connection Manager for the component. This will be the file to which you write the data from upstream.

Next, you add back the input:

IDTSInput100 input = ComponentMetaData.InputCollection.New();
input.Name = “Component Input”;
input.Description = “This is what we see from the upstream
component”;

This is what the upstream component will connect to, and through which you will receive the data from the previous component. You need to ensure that the IDTSInput100 object of the component remains in the execution plan, regardless of whether it is attached, by setting the HasSideEffects property to true. This means that at runtime, the SSIS execution engine is smart enough to “prune” from the package any components that are not actually doing any work. You need to explicitly tell SSIS that this component is doing work (external file writes) by setting this property:

input.HasSideEffects = true;

After completing the ProvideComponentProperties method, you can move on to the AcquireConnections method (ProfSSISDestAdapter.cs). This method is not really any different from the AcquireConnections method you saw in the Source adapter; the method is shown in full but without being described in detail. If you need line-by-line details about what’s happening, you can refer back to the Source adapter. This method accomplishes the following tasks:

  • Checks whether the user has supplied a Connection Manager to the component.
  • Checks whether the Connection Manager is a FILE Connection Manager.
  • Ensures that the FILE Connection Manager has a FileUsageType property value of DTSFileConnectionUsageType.CreateFile. (This is different from the Source, which required an existing file.)
  • Gets the filename from the Connection Manager.

public override void AcquireConnections(object transaction)
{
bool pbCancel = false;
if (ComponentMetaData.RuntimeConnectionCollection[“File To
Write”].ConnectionManager != null)
{
ConnectionManager cm =
Microsoft.SqlServer.Dts.Runtime.DtsConvert.GetWrapper(
ComponentMetaData.RuntimeConnectionCollection[“File To Write”]
.ConnectionManager);
if cm.CreationName != “FILE”)
{
ComponentMetaData.FireError(0, “Acquire Connections”, “The Connection
Manager is not a FILE Connection Manager”, “”, 0, out pbCancel);
throw new Exception(“The Connection Manager is not a FILE Connection
Manager”);
}
else
{_
fil = (DTSFileConnectionUsageType)cm.Properties[“FileUsageType”]
.GetValue(cm);
if (_fil != DTSFileConnectionUsageType.CreateFile)
{
ComponentMetaData.FireError(0, “Acquire Connections”,
“The type of FILE connection manager must be Create File”, “”,
0, out pbCancel);
throw new Exception(“The type of FILE connection manager must be
Create File”);
}
else
{
_filename = ComponentMetaData.RuntimeConnectionCollection
[“File To Read”].ConnectionManager.AcquireConnection(transaction)
.ToString();
if (_filename == null || _filename.Length == 0)
{
ComponentMetaData.FireError(0, “Acquire Connections”, “Nothing
returned when grabbing the filename”, “”, 0, out pbCancel);
throw new Exception(“Nothing returned when grabbing the filename”);
}
}
}
}
}

A lot of ground is covered in the AcquireConnections method. Much of this code is covered again in the Validate method, which you will visit now. The Validate method also checks whether the input to the component is correct; if it isn’t, you try to fix what is wrong by calling ReinitializeMetaData. Here is the Validate method (ProfSSISDestAdapter.cs):

[CLSCompliant(false)]
public override DTSValidationStatus Validate()
{
bool pbCancel = false;
if (ComponentMetaData.OutputCollection.Count != 0)
{
ComponentMetaData.FireError(0, ComponentMetaData.Name,
“Unexpected Output
Found. Destination components do not support outputs.”, “”,
0, out pbCancel);
return DTSValidationStatus.VS_ISCORRUPT;
}
if (ComponentMetaData.RuntimeConnectionCollection[“File To Write”]
.ConnectionManager == null)
{
ComponentMetaData.FireError(0, “Validate”, “No Connection Manager
returned”,
“”, 0, out pbCancel);
return DTSValidationStatus.VS_ISCORRUPT;
}
if (ComponentMetaData.AreInputColumnsValid == false)
{
ComponentMetaData.InputCollection[“Component Input”]
InputColumnCollection.RemoveAll();
return DTSValidationStatus.VS_NEEDSNEWMETADATA;
}
return base.Validate();
}

The first check in the method ensures that the component has no outputs:

bool pbCancel = false;
if (ComponentMetaData.OutputCollection.Count != 0)
{
ComponentMetaData.FireError(0, ComponentMetaData.Name, “Unexpected
Output found.
Destination components do not support outputs.”, “”, 0, out
pbCancel);
return DTSValidationStatus.VS_ISCORRUPT;
}

You now want to ensure that the user specified a Connection Manager. Remember that you are only validating the fact that a Connection Manager is specified, not whether it is a valid type. The extensive checking of the Connection Manager is done in AcquireConnections().

if (ComponentMetaData.RuntimeConnectionCollection[“File To
Write”].ConnectionManager == null)
{
ComponentMetaData.FireError(0, “Validate”, “No Connection Manager
returned”, “”,
0, out pbCancel);
return DTSValidationStatus.VS_ISCORRUPT;
}

The final thing you do in this method is to check whether the input columns are valid. Valid in this instance means that the columns in the input collection reference existing columns in the upstream component. If this is not the case, you call the trusty ReinitializeMetaData method.

if (ComponentMetaData.AreInputColumnsValid == false)
{
ComponentMetaData.InputCollection[“Component Input”]
.InputColumnCollection.RemoveAll();
return DTSValidationStatus.VS_NEEDSNEWMETADATA;

The return value DTSValidationStatus.VS_NEEDSNEWMETADATA means that the component will now call ReinitializeMetaData to try to sort out the problems with the component. Here is that method in full:

public override void ReinitializeMetaData()
{
IDTSInput100 _profinput =
ComponentMetaData.InputCollection[“Component Input”];
_profinput.InputColumnCollection.RemoveAll();
IDTSVirtualInput100 vInput = _profinput.GetVirtualInput();
foreach (IDTSVirtualInputColumn100 vCol in
vInput.VirtualInputColumnCollection)
{
this.SetUsageType(_profinput.ID, vInput, vCol.LineageID,
DTSUsageType.UT_READONLY);
}
}

NOTE: Notice that the columns are blown away in ReinitializeMetaData and built again from scratch. A better solution is to test what the invalid columns are and try to fix them. If you cannot fix them, you could remove them, and then the user could reselect at leisure. Tutorials Online has an example of doing this.

The IDTSVirtualInput and IDTSVirtualInputColumnCollection in this component need a little explanation. There is a subtle difference between these two objects and their input equivalents. The “virtual” objects are what your component could have as inputs — that is to say, they are upstream inputs and columns that present themselves as available to your component. The inputs themselves are what you have chosen for your component to have as inputs from the virtual object. In the ReinitializeMetaData method, you start by removing all existing input columns:

IDTSInput100 _profinput =
ComponentMetaData.InputCollection[“Component Input”];
_profinput.InputColumnCollection.RemoveAll();

You then get a reference to the input’s virtual input:

IDTSVirtualInput100 vInput = _profinput.GetVirtualInput();

Now that you have the virtual input, you can add an input column to the component for every virtual input column you find:

foreach (IDTSVirtualInputColumn100 vCol in
vInput.VirtualInputColumnCollection)
{
this.SetUsageType(_profinput.ID, vInput, vCol.LineageID,
DTSUsageType.UT_READONLY);
}

The SetUsageType method simply adds an input column to the input column collection of the component, or removes it depending on your UsageType value. When a user adds a connector from an upstream component that contains its output to this component and attaches it to this component’s input, the OnInputAttached method is called. This method has been overridden in the component herein:

public override void OnInputPathAttached(int inputID)
{I
DTSInput100 input =
ComponentMetaData.InputCollection.GetObjectByID(inputID);
IDTSVirtualInput100 vInput = input.GetVirtualInput();
foreach (IDTSVirtualInputColumn100 vCol in
vInput.VirtualInputColumnCollection)
{t
his.SetUsageType(inputID, vInput, vCol.LineageID,
DTSUsageType.UT_READONLY);
}
}

This method is the same as the ReinitializeMetaData method except that you don’t need to remove the input columns from the collection. This is because if the input is not mapped to the output of an upstream component, there can be no input columns. 

That completes the design-time methods for your component. You can now move on to look at the runtime methods. You are going to be looking at only two methods: PreExecute and ProcessInput.

PreExecute is executed only once in this component (ProfSSISDestAdapter.cs), so you want to do as much preparation work as you can in this method. It is also the first opportunity in the component to access the Buffer Manager, which contains the columns. In this component, you use it for two things: getting the information about the component’s input columns and storing essential details about them:

public override void PreExecute()
{
IDTSInput100 input = ComponentMetaData.InputCollection[“Component
Input”];
foreach (IDTSInputColumn100 inCol in input.InputColumnCollection)
{
ColumnInfo ci = new ColumnInfo();
ciBufferColumnIndex =
BufferManager.FindColumnByLineageID(input.Buffer,
inCol.LineageID);
ci.ColumnName = inCol.Name;
_columnInfos.Add(ci);
}/
/ Open the file
_fs = new FileStream(_filename, FileMode.OpenOrCreate,
FileAccess.Write);
_sw = new StreamWriter(_fs);
}

First you get a reference to the component’s input:

IDTSInput100 input = ComponentMetaData.InputCollection[“Component
Input”];

Then you loop through the input’s InputColumnCollection:

foreach (IDTSInputColumn100 inCol in input.InputColumnCollection)
{

For each input column you find, you need to create a new instance of the ColumnInfo struct. You then assign to the struct values you can retrieve from the input column itself and the Buffer Manager. You assign these values to the struct and finally add them to the array that is holding all the ColumnInfo objects:

ColumnInfo ci = new ColumnInfo();
ci.BufferColumnIndex =
BufferManager.FindColumnByLineageID(input.Buffer,
inCol.LineageID);
ci.ColumnName = inCol.Name;
_columnInfos.Add(ci);

Doing things this way will enable you to move more quickly through the ProcessInput method. The last thing you do in the PreExecute method is get a reference to the file to which you want to write:

_fs = new FileStream(_filename, FileMode.OpenOrCreate,
FileAccess.Write);
_sw = new StreamWriter(_fs);

You will use this in the next method, ProcessInput (ProfSSISDestAdapter.cs). ProcessInput is where you are going to keep reading the rows that are coming from the upstream component. While there are rows, you write those values to a file. This is a very simplistic view of what needs to be done, so you should have a look at how to make that happen.

public override void ProcessInput(int inputID, PipelineBuffer buffer)
{
while (buffer.NextRow())
{
_sw.WriteLine(“<START>”);
for (int i = 0; i < _columnInfos.Count; i++)
{
ColumnInfo ci = (ColumnInfo)_columnInfos[i];
if (buffer.IsNull(ci.BufferColumnIndex))
{
_sw.WriteLine(ci.ColumnName + “:”);
}
else
{
_sw.WriteLine(ci.ColumnName + “:” +
bffer[ci.BufferColumnIndex].ToString());
}
}
_sw.WriteLine(“<END>”);
}
if (buffer.EndOfRowset) _sw.Close();
}

The first thing you do is check whether there are still rows in the buffer:

while (buffer.NextRow())
{

You now need to loop through the array that is holding the collection of ColumnInfo objects that were populated in the PreExecute method:

for (int i = 0; i < _columnInfos.Count; i++)

For each iteration, you create a new instance of the ColumnInfo object:

You now need to retrieve from the buffer object the value of the column whose index you will pass in from the ColumnInfo object. If the value is not null, you write the value of the column and the column name to the text file. If the value is null, you write just the column name to the text file. Again, because you took the time to store these details in a ColumnInfo object earlier, retrieving these properties is easy.

if (buffer.IsNull(ci.BufferColumnIndex))
{
_sw.WriteLine(ci.ColumnName + “:”);
}
else
{
_sw.WriteLine(ci.ColumnName + “:” +
buffer[ci.BufferColumnIndex].ToString());
}

Finally, you check whether the upstream component has called SetEndOfRowset; if so, you close the file stream:

if (buffer.EndOfRowset) _sw.Close();

Having concluded your look at the Destination adapter, you are ready to learn how you get SSIS to recognize your components and what properties you need to assign to them.

0 Responses on Developing the Destination Adapter"

Leave a Message

Your email address will not be published. Required fields are marked *

Copy Rights Reserved © Mindmajix.com All rights reserved. Disclaimer.
Course Adviser

Fill your details, course adviser will reach you.