SSIS Script Transformation - Performing WebClient Async Operation

Question

I'm receiving 6 unique URLs for different domains/hosts, parsing the XML they return, and inserting into a table. I had this working fine except it was doing each server synchronously (one at a time). I need it to reach out to all six servers at the same time.

I changed my WebClient calls to use the Async method then added a while statement to check if all the WebClient calls have returned, however the SSIS package seems to get cancelled after the while statement completes, whether I have the call to SetEndOfRowSet or not. No errors are shown within SSIS logging (even when selecting all the log options). It's says 'Cancelled' in the Output Window and there's a brief console window that pops up when it happens but I can't catch it.

If I don't have the while statement in the SSIS package, the package continues past my script transformation and no rows are output from it. The script transformation is set to None on SynchronousInputID.

Here's the SSIS C# Script Transformation code:

/* Microsoft SQL Server Integration Services Script Component
*  Write scripts using Microsoft Visual C# 2008.
*  ScriptMain is the entry point class of the script.*/

using System;
using System.Data;
using Microsoft.SqlServer.Dts.Pipeline.Wrapper;
using Microsoft.SqlServer.Dts.Runtime.Wrapper;
using System.Net;
using System.Diagnostics;
using System.Xml;

[Microsoft.SqlServer.Dts.Pipeline.SSISScriptComponentEntryPointAttribute]
public class ScriptMain : UserComponent

{
    private int CountComplete = 0;

public override void PreExecute()
{
    base.PreExecute();
}

public override void PostExecute()
{
    base.PostExecute();
}

public override void Input0_ProcessInput(Input0Buffer Buffer)
{
    while(Buffer.NextRow())
    {
        Input0_ProcessInputRow(Buffer);
    }

    if (Buffer.EndOfRowset())
    {
        while (CountComplete < Variables.intRows)
        {
            System.Threading.Thread.Sleep(500);
        }
        Output0Buffer.SetEndOfRowset();
    }
}

public override void Input0_ProcessInputRow(Input0Buffer Row)
{
    string DNS_NA = Row.DNSNA;
    long SERVER = Convert.ToInt64(Row.SERSYSNR);

    // Ignore server/certificate mismatch...
    ServicePointManager.ServerCertificateValidationCallback += (sender, certificate, chain, sslPolicyErrors) => true;

    // Declare variable that will hold the xml document received by the API
    string xmlDoc = String.Empty;

    // Get the URI from the variable
    string url = @"https://" + DNS_NA + Variables.strURLSuffix;

    // Time the request...
    Stopwatch timer = Stopwatch.StartNew();
    //Create a Web Client
    WebClient client = new WebClient();
    client.DownloadStringCompleted += (sender, e) =>
    {
        timer.Stop();
        processRow(e.Result, SERVER, Convert.ToInt32(timer.ElapsedMilliseconds), Convert.ToInt32(Row.UPTIMQY), Output0Buffer);
        CountComplete += 1;
    };
    Uri uri = new Uri(url);
    client.DownloadStringAsync(uri);
}

private void processRow(string xmlDoc, long SERVER, int FRAPI_TIME, int PRE_UPTIME, Output0Buffer buffer)
{
    try
    {
        XmlDocument xDoc = new XmlDocument();

        xDoc.LoadXml(xmlDoc);

        bool NORESPONSE = false; // If we made it this far, we got a response, right?
        short CPU = Convert.ToInt16(xDoc.SelectSingleNode("//server").ChildNodes[1].InnerText);
        int TOT_MEM = Convert.ToInt32(xDoc.SelectSingleNode("//server").ChildNodes[2].InnerText) / (1024 * 1024);
        int USE_MEM = Convert.ToInt32(xDoc.SelectSingleNode("//server").ChildNodes[3].InnerText) / (1024 * 1024);
        int UPTIME = Convert.ToInt32(xDoc.SelectSingleNode("//server").ChildNodes[4].InnerText) / 1000 / 60;
        short REQUESTS = Convert.ToInt16(xDoc.SelectSingleNode("//server").ChildNodes[5].InnerText);
        bool RESET = (UPTIME < PRE_UPTIME) ? true : false;


        foreach (XmlNode xNode in xDoc.SelectNodes("//server/detail/request"))
        {
            buffer.AddRow();
            buffer.SERSYSNR = SERVER;
            buffer.CPUUSEQY = CPU;
            buffer.TOTMEMQY = TOT_MEM;
            buffer.USEMEMQY = USE_MEM;
            buffer.UPTIMQY = UPTIME;
            buffer.REQCNTQY = REQUESTS;
            buffer.REQMSQY = FRAPI_TIME;
            buffer.SERNONRSPIR = NORESPONSE;
            buffer.SERRSTIR = RESET;
            buffer.REQSYSNR = Convert.ToInt32(xNode.ChildNodes[0].InnerText);
            buffer.REQIPTE = xNode.ChildNodes[1].InnerText;
            buffer.REQURLTE = xNode.ChildNodes[2].InnerText;
            buffer.REQRUNTMQY = Convert.ToInt32(xNode.ChildNodes[3].InnerText);

        }
    }
    catch
    {
        // Hopefully we didn't throw an exception inside the foreach where a row has already been added...
        buffer.AddRow();
        buffer.SERSYSNR = SERVER;
        buffer.SERNONRSPIR = true;
    }

}

   // public override void CreateNewOutputRows()
   // {
        /*
          Add rows by calling the AddRow method on the member variable named "<Output Name>Buffer".
          For example, call MyOutputBuffer.AddRow() if your output was named "MyOutput".
        */
    //}

}

I can use a conditional split on the DNS_NA and make x number of script components and then UNION ALL them, but each time a new server is added to the list I'll have to edit the SSIS package. Not ideal.

Answer 1

Related Connect bug report: https://connect.microsoft.com/SQLServer/feedback/details/367692/ssis-detects-end-of-data-flow-components-in-the-wrong-way

Splitting into 6 number of script tasks was the only solution that worked. I did fix the SSIS crash when using the DownloadStringSync by switching to a threaded setup and calling just DownloadString. I had to make sure I locked Output0Buffer when processing the results which stumped me for a bit. I ended up using a list to store the threads and looping over them and calling Thread.Join() after I checked for Buffer.EndOfRowSet() in ProcessInputBuffer.

However that solution still wouldn't forward the rows until all the requests had completed which meant if one server ran for a long period of time, the results weren't put into the database until all had returned.