Added support of JSON protocol for Binary ETW reader and writer

Source/AssemblyInfo.cs

@@ -5,5 +5,5 @@ using System.Reflection;
 [assembly: AssemblyCompany("MS Open Tech")]
 [assembly: AssemblyProduct("Tx (LINQ to Logs and Traces)")]
 [assembly: AssemblyCopyright("Copyright © MS Open Tech 2012")]
-[assembly: AssemblyVersion("1.0.50825.0")]
-[assembly: AssemblyFileVersion("1.0.50825.0")]
+[assembly: AssemblyVersion("1.0.50917.0")]
+[assembly: AssemblyFileVersion("1.0.50917.0")]

Source/Tx.Bond/BondEtwObserver.cs

@@ -6,7 +6,9 @@ namespace Tx.Bond
     using System.Collections.Concurrent;
     using System.Collections.Generic;
     using System.Linq;
+    using System.Text;
     using System.Threading;
+    using System.Web.Script.Serialization;
 
     using global::Bond;
     using global::Bond.IO.Safe;
@@ -16,7 +18,13 @@ namespace Tx.Bond
 
     public sealed class BondEtwObserver : IObserver<object>, IDisposable
     {
-        private IDictionary<Type, string> manifestMap;
+        private readonly IDictionary<Type, string> manifestMap;
+
+        private readonly OutputBuffer outputBuffer = new OutputBuffer();
+
+        private readonly CompactBinaryWriter<OutputBuffer> writer;
+
+        private readonly JavaScriptSerializer javaScriptSerializer = new JavaScriptSerializer();
 
         private TimeSpan interval = TimeSpan.FromMinutes(20);
 
@@ -26,17 +34,23 @@ namespace Tx.Bond
 
         private ConcurrentBag<BondTypeInfo> knownManifest;
 
-        private readonly OutputBuffer outputBuffer = new OutputBuffer();
-
-        private readonly CompactBinaryWriter<OutputBuffer> writer;
-
+        /// <summary>
+        /// Initializes a new instance of the <see cref="BondEtwObserver"/> class.
+        /// </summary>
         public BondEtwObserver()
         {
             this.writer = new CompactBinaryWriter<OutputBuffer>(this.outputBuffer);
         }
 
+        /// <summary>
+        /// Initializes a new instance of the <see cref="BondEtwObserver"/> class.
+        /// </summary>
+        /// <param name="manifestMap">The map of manifests per types.</param>
+        /// <param name="interval">The time period between writing manifests into ETW.</param>
+        /// <exception cref="System.ArgumentNullException">manifestMap</exception>
+        /// <exception cref="System.ArgumentOutOfRangeException">interval</exception>
         public BondEtwObserver(
-            IDictionary<Type, string> manifestMap, 
+            IDictionary<Type, string> manifestMap,
             TimeSpan interval)
         {
             if (manifestMap == null)
@@ -56,7 +70,7 @@ namespace Tx.Bond
 
             this.Initialize();
         }
-        
+
         public void Initialize()
         {
             if (this.bondTypeMap == null)
@@ -82,6 +96,10 @@ namespace Tx.Bond
             }
         }
 
+        /// <summary>
+        /// Provides the observer with new data.
+        /// </summary>
+        /// <param name="value">The current notification information.</param>
         public void OnNext(object value)
         {
             if (value == null)
@@ -96,21 +114,34 @@ namespace Tx.Bond
             BondTypeInfo manifestData;
             if (!this.bondTypeMap.TryGetValue(type, out manifestData))
             {
-                if (!type.IsBondType())
+                string manifest;
+
+                if (type.IsBondStruct())
                 {
-                    throw new NotSupportedException();
+                    manifest = type.TryGetManifestData();
+
+                    manifestData = new BondTypeInfo
+                    {
+                        ManifestId = type.GetBondManifestIdentifier(),
+                        ManifestData = manifest,
+                        Serializer = new Serializer<CompactBinaryWriter<OutputBuffer>>(type)
+                    };
+
+                    this.bondTypeMap.Add(type, manifestData);
                 }
-
-                var manifest = type.TryGetManifestData();
-
-                manifestData = new BondTypeInfo
+                else
                 {
-                    ManifestId = type.GetBondManifestIdentifier(),
-                    ManifestData = manifest,
-                    Serializer = new Serializer<CompactBinaryWriter<OutputBuffer>>(type)
-                };
+                    manifest = null;
 
-                this.bondTypeMap.Add(type, manifestData);
+                    manifestData = new BondTypeInfo
+                    {
+                        ManifestId = type.GetBondManifestIdentifier(),
+                        ManifestData = null,
+                        Serializer = null
+                    };
+
+                    this.bondTypeMap.Add(type, manifestData);
+                }
 
                 if (manifest != null)
                 {
@@ -123,21 +154,43 @@ namespace Tx.Bond
                 }
             }
 
-            this.outputBuffer.Position = 0;
-
-            manifestData.Serializer.Serialize(value, this.writer);
-
             var now = DateTime.UtcNow;
-            BinaryEventSource.Log.Write(now, now, BondProtocol.CompactBinaryV1, "Tx.Bond", this.outputBuffer.Data.ToByteArray(), manifestData.ManifestId);
+
+            if (manifestData.Serializer != null)
+            {
+                this.outputBuffer.Position = 0;
+
+                manifestData.Serializer.Serialize(value, this.writer);
+
+                BinaryEventSource.Log.Write(now, now, BondProtocol.CompactBinaryV1, @"Tx.Bond", this.outputBuffer.Data.ToByteArray(), manifestData.ManifestId);
+            }
+            else
+            {
+                var json = this.javaScriptSerializer.Serialize(value);
+                BinaryEventSource.Log.Write(
+                    now,
+                    now,
+                    "JSON",
+                    @"Tx.Bond",
+                    Encoding.UTF8.GetBytes(json),
+                    manifestData.ManifestId);
+            }
         }
 
+        /// <summary>
+        /// Notifies the observer that the provider has finished sending push-based notifications.
+        /// </summary>
         public void OnCompleted()
         {
         }
 
+        /// <summary>
+        /// Notifies the observer that the provider has experienced an error condition.
+        /// </summary>
+        /// <param name="error">An object that provides additional information about the error.</param>
         public void OnError(Exception error)
         {
-            throw error;
+            BinaryEventSource.Log.Error(error.ToString());
         }
 
         private void InitializeTimer()
@@ -163,11 +216,15 @@ namespace Tx.Bond
                         manfiestInfo.ManifestData);
                 }
             }
+            catch(Exception error)
+            {
+                BinaryEventSource.Log.Error(error.ToString());
+            }
             finally
             {
                 if (this.logManifestTimer != null)
                 {
-                    this.logManifestTimer.Change(this.interval.Ticks, Timeout.Infinite);
+                    this.logManifestTimer.Change((int)this.interval.TotalMilliseconds, Timeout.Infinite);
                 }
             }
         }

Source/Tx.Bond/BondIdentifierHelpers.cs

@@ -4,7 +4,6 @@ namespace Tx.Bond
 {
     using System;
     using System.Linq;
-    using System.Reflection;
     using System.Runtime.InteropServices;
     using System.Security.Cryptography;
     using System.Text;
@@ -31,19 +30,14 @@ namespace Tx.Bond
             (byte) 251
         };
 
-        public static string GetBondManifestIdentifier(this object bondSerializable)
+        public static string GetBondManifestIdentifier(this object instance)
         {
-            if (bondSerializable == null)
+            if (instance == null)
             {
-                throw new ArgumentNullException("bondSerializable");
+                throw new ArgumentNullException("instance");
             }
 
-            var type = bondSerializable.GetType();
-
-            if (!type.IsBondType())
-            {
-                throw new NotSupportedException();
-            }
+            var type = instance.GetType();
 
             var manifestId = type.GetBondManifestIdentifier();
 
@@ -57,11 +51,6 @@ namespace Tx.Bond
                 throw new ArgumentNullException("type");
             }
 
-            if (!type.IsBondType())
-            {
-                return null;
-            }
-
             var bondMapAttribute = ((GuidAttribute[])type.GetCustomAttributes(typeof(GuidAttribute), false))
                 .FirstOrDefault();
 

Source/Tx.Bond/GeneralPartitionableTypeMap.cs

@@ -0,0 +1,153 @@
+// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. See License.txt in the project root for license information.
+
+namespace Tx.Bond
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Reactive;
+    using System.Text;
+    using System.Web.Script.Serialization;
+
+    using global::Bond;
+    using global::Bond.IO.Safe;
+    using global::Bond.Protocols;
+
+    using Tx.Binary;
+
+    public sealed class GeneralPartitionableTypeMap : IPartitionableTypeMap<BinaryEnvelope, string>
+    {
+        private static readonly IEqualityComparer<string> comparer = StringComparer.OrdinalIgnoreCase;
+
+        // Objects returned by transform call should not be null and be of System.Object type.
+        private static readonly object defaultInstance = new { };
+
+        private static readonly JavaScriptSerializer javaScriptSerializer = new JavaScriptSerializer();
+
+        private readonly Dictionary<Type, Func<BinaryEnvelope, object>> transforms = new Dictionary<Type, Func<BinaryEnvelope, object>>();
+
+        public IEqualityComparer<string> Comparer
+        {
+            get { return comparer; }
+        }
+
+        public string GetTypeKey(Type outputType)
+        {
+            string manifestId;
+
+            try
+            {
+                manifestId = outputType.GetBondManifestIdentifier();
+            }
+            catch
+            {
+                manifestId = string.Empty;
+            }
+
+            return manifestId;
+        }
+
+        public Func<BinaryEnvelope, object> GetTransform(Type outputType)
+        {
+            Func<BinaryEnvelope, object> transform;
+            this.transforms.TryGetValue(outputType, out transform);
+
+            if (transform != null)
+            {
+                return transform;
+            }
+
+            Func<BinaryEnvelope, object> jsonDeserializer = e => DeserializeJson(e.EventPayload, outputType);
+
+            var deserializerMap = new Dictionary<string, Func<BinaryEnvelope, object>>(StringComparer.OrdinalIgnoreCase)
+            {
+                { "JSON", jsonDeserializer }
+            };
+
+            if (outputType.IsBondStruct())
+            {
+                var deserializer = new Deserializer<CompactBinaryReader<InputBuffer>>(outputType);
+
+                deserializerMap.Add("BOND_V1", e => DeserializeCompactBinary(1, e.EventPayload, deserializer));
+                deserializerMap.Add("BOND", e => DeserializeCompactBinary(2, e.EventPayload, deserializer));
+            }
+
+            transform = e => Transform(e, deserializerMap);
+
+            this.transforms.Add(outputType, transform);
+
+            return transform;
+        }
+
+        public Func<BinaryEnvelope, DateTimeOffset> TimeFunction
+        {
+            get
+            {
+                return GetTime;
+            }
+        }
+
+        public string GetInputKey(BinaryEnvelope envelope)
+        {
+            return envelope.PayloadId;
+        }
+
+        private static DateTimeOffset GetTime(BinaryEnvelope envelope)
+        {
+            var time = DateTimeOffset.FromFileTime(envelope.ReceiveFileTimeUtc);
+
+            return time;
+        }
+
+        private static object Transform(
+            BinaryEnvelope envelope,
+            IDictionary<string, Func<BinaryEnvelope, object>> deserializerMap)
+        {
+            if (envelope.EventPayload == null)
+            {
+                return defaultInstance;
+            }
+
+            object deserializedObject;
+
+            try
+            {
+                Func<BinaryEnvelope, object> transform;
+                if (deserializerMap.TryGetValue(envelope.Protocol, out transform))
+                {
+                    deserializedObject = transform(envelope) ?? defaultInstance;
+                }
+                else
+                {
+                    deserializedObject = defaultInstance;
+                }
+            }
+            catch
+            {
+                deserializedObject = defaultInstance;
+            }
+
+            return deserializedObject;
+        }
+
+        private static object DeserializeCompactBinary(
+            ushort version,
+            byte[] data,
+            Deserializer<CompactBinaryReader<InputBuffer>> deserializer)
+        {
+            var inputStream = new InputBuffer(data);
+
+            var reader = new CompactBinaryReader<InputBuffer>(inputStream, version);
+
+            var outputObject = deserializer.Deserialize(reader);
+
+            return outputObject;
+        }
+
+        private static object DeserializeJson(byte[] data, Type outputType)
+        {
+            var json = Encoding.UTF8.GetString(data);
+
+            return javaScriptSerializer.Deserialize(json, outputType);
+        }
+    }
+}

Source/Tx.Bond/Tx.Bond.csproj

@@ -50,6 +50,8 @@
     <Reference Include="System.Reactive.PlatformServices">
       <HintPath>$(CPReferencePath)\System.Reactive.PlatformServices.dll</HintPath>
     </Reference>
+    <Reference Include="System.Web" />
+    <Reference Include="System.Web.Extensions" />
   </ItemGroup>
   <ItemGroup>
     <Compile Include="..\AssemblyInfo.cs">
@@ -68,6 +70,7 @@
     <Compile Include="ByteArrayHelper.cs" />
     <Compile Include="EtwExtensions.cs" />
     <Compile Include="EventManifest.cs" />
+    <Compile Include="GeneralPartitionableTypeMap.cs" />
     <Compile Include="ITypeProvider.cs" />
     <Compile Include="InvalidBondTypeException.cs" />
     <Compile Include="StringHelper.cs" />

Test/UnitTests/BinaryEtw/BondEtwObserverTests.cs

@@ -0,0 +1,44 @@
+namespace Tests.Tx.BinaryEtw
+{
+    using System;
+    using System.Collections.Generic;
+
+    using global::Tx.Bond;
+
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+    [TestClass]
+    public class BondEtwObserverTests
+    {
+        [TestMethod]
+        public void WriteToBinaryEtw()
+        {
+            using (var observer = new BondEtwObserver())
+            {
+                observer.OnNext(new GeneralPartitionableTypeMapTests.TestBondClass{ EventId = "A" });
+                observer.OnNext("A");
+                observer.OnNext(null);
+
+                observer.OnCompleted();
+            }
+        }
+
+        [TestMethod]
+        public void WriteToBinaryEtw_2()
+        {
+            using (var observer = new BondEtwObserver(
+                new Dictionary<Type, string>
+                {
+                    { typeof(GeneralPartitionableTypeMapTests.TestBondClass), "Manifest" }
+                }, 
+                TimeSpan.FromMinutes(1)))
+            {
+                observer.OnNext(new GeneralPartitionableTypeMapTests.TestBondClass { EventId = "A" });
+                observer.OnNext("A");
+                observer.OnNext(null);
+
+                observer.OnCompleted();
+            }
+        }
+    }
+}

Test/UnitTests/BinaryEtw/GeneralPartitionableTypeMapTests.cs

@@ -0,0 +1,114 @@
+namespace Tests.Tx.BinaryEtw
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Reactive;
+    using System.Reactive.Linq;
+    using System.Runtime.InteropServices;
+    using System.Text;
+
+    using global::Tx.Binary;
+    using global::Tx.Bond;
+
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+    [TestClass]
+    public class GeneralPartitionableTypeMapTests
+    {
+        [TestMethod]
+        public void DeserializeJson()
+        {
+            var typeMap = new GeneralPartitionableTypeMap();
+
+            var typekey = typeMap.GetTypeKey(typeof(string));
+            var envelope = new BinaryEnvelope
+            {
+                EventPayload = Encoding.UTF8.GetBytes(@"""A"""),
+                PayloadId = "daf0be6e-da1e-5a6a-0d49-69782745c885",
+                Protocol = "JSON"
+            };
+
+            var typekey2 = typeMap.GetInputKey(envelope);
+            Assert.AreEqual(typekey, typekey2);
+
+            var result = typeMap.GetTransform(typeof(string))(envelope);
+            Assert.AreEqual("A", result);
+        }
+
+        [TestMethod]
+        public void DeserializeBondV1()
+        {
+            var typeMap = new GeneralPartitionableTypeMap();
+
+            var typekey = typeMap.GetTypeKey(typeof(TestBondClass));
+
+            var envelope = new BinaryEnvelope
+            {
+                EventPayload = new byte[] { 41, 1, 65, 0 },
+                PayloadId = "daf0be6e-da1e-5a6a-0d49-69782745c886",
+                Protocol = "BOND_V1"
+            };
+
+            var typekey2 = typeMap.GetInputKey(envelope);
+            Assert.AreEqual(typekey, typekey2);
+
+            var result = typeMap.GetTransform(typeof(TestBondClass))(envelope);
+        }
+
+        [TestMethod]
+        public void DeserializeMixedStream()
+        {
+            var input = new[]
+            {
+                new SimpleEnvelope("JSON", Guid.Parse("daf0be6e-da1e-5a6a-0d49-69782745c885"), Encoding.UTF8.GetBytes(@"""A""")), 
+                new SimpleEnvelope("JSON", Guid.Parse("daf0be6e-da1e-5a6a-0d49-69782745c885"), new byte[0]), 
+                new SimpleEnvelope("BOND_V1", new Guid("daf0be6e-da1e-5a6a-0d49-69782745c886"), new byte[] { 41, 1, 65, 0 }), 
+            };
+
+            IEnumerable<TestBondClass> testBondClassCollection;
+            IEnumerable<string> stringCollection;
+
+            using (var playback = new Playback())
+            {
+                ((IPlaybackConfiguration)playback).AddInput(
+                    () => input.ToObservable(),
+                    typeof(GeneralPartitionableTypeMap));
+
+                var testBondClassStream = playback.GetObservable<TestBondClass>();
+                var stringStream = playback.GetObservable<string>();
+
+                testBondClassCollection = playback.BufferOutput(testBondClassStream);
+                stringCollection = playback.BufferOutput(stringStream);
+
+                playback.Run();
+            }
+
+            var bondObjects = testBondClassCollection.ToArray();
+            Assert.IsNotNull(bondObjects);
+            Assert.AreEqual(1, bondObjects.Length);
+
+            var strings = stringCollection.ToArray();
+            Assert.IsNotNull(strings);
+            Assert.AreEqual(1, strings.Length);
+        }
+
+        private class SimpleEnvelope : BinaryEnvelope
+        {
+            public SimpleEnvelope(string protocol, Guid manifestId, byte[] data)
+            {
+                this.Protocol = protocol;
+                this.PayloadId = manifestId.ToString();
+                this.EventPayload = data;
+            }
+        }
+
+        [global::Bond.Schema]
+        [Guid("daf0be6e-da1e-5a6a-0d49-69782745c886")]
+        public partial class TestBondClass
+        {
+            [global::Bond.Id(1)]
+            public string EventId { get; set; }
+        }
+    }
+}

Test/UnitTests/Tests.Tx.csproj

@@ -25,6 +25,16 @@
     <AssemblyOriginatorKeyFile>..\..\Source\key.snk</AssemblyOriginatorKeyFile>
   </PropertyGroup>
   <ItemGroup>
+    <Reference Include="Bond">
+      <HintPath>$(CPReferencePath)\Bond.dll</HintPath>
+    </Reference>
+    <Reference Include="Bond.Attributes">
+      <HintPath>$(CPReferencePath)\Bond.Attributes.dll</HintPath>
+    </Reference>
+    <Reference Include="Microsoft.Diagnostics.Tracing.EventSource">
+      <SpecificVersion>False</SpecificVersion>
+      <HintPath>$(CPReferencePath)\Microsoft.Diagnostics.Tracing.EventSource.dll</HintPath>
+    </Reference>
     <Reference Include="Microsoft.Reactive.Testing">
       <HintPath>$(CPReferencePath)\Microsoft.Reactive.Testing.dll</HintPath>
     </Reference>
@@ -56,6 +66,8 @@
     <Compile Include="..\..\Generated\Microsoft_Windows_Kernel_Process.cs">
       <Link>Microsoft_Windows_Kernel_Process.cs</Link>
     </Compile>
+    <Compile Include="BinaryEtw\BondEtwObserverTests.cs" />
+    <Compile Include="BinaryEtw\GeneralPartitionableTypeMapTests.cs" />
     <Compile Include="DemultiplexorTest.cs" />
     <Compile Include="EtwDeserializerTest.cs" />
     <Compile Include="EtwGeneration.cs" />
@@ -99,6 +111,10 @@
     <Folder Include="Properties\" />
   </ItemGroup>
   <ItemGroup>
+    <ProjectReference Include="..\..\Source\Tx.Bond\Tx.Bond.csproj">
+      <Project>{6dce4a40-4946-41b0-abe5-ce1700b598e9}</Project>
+      <Name>Tx.Bond</Name>
+    </ProjectReference>
     <ProjectReference Include="..\..\Source\Tx.Core\Tx.Core.csproj">
       <Project>{c5cc33b0-1684-4dd4-83a5-5da4a9a25a7f}</Project>
       <Name>Tx.Core</Name>

Test/UnitTests/Tests.Tx.sln

@@ -1,12 +1,16 @@
 
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio 2012
+# Visual Studio 2013
+VisualStudioVersion = 12.0.31101.0
+MinimumVisualStudioVersion = 10.0.40219.1
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tests.Tx", "Tests.Tx.csproj", "{E88FF544-1342-4ADC-B88A-1BDCE0057DA6}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tx.Core", "..\..\Source\Tx.Core\Tx.Core.csproj", "{C5CC33B0-1684-4DD4-83A5-5DA4A9A25A7F}"
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tx.Windows", "..\..\Source\Tx.Windows\Tx.Windows.csproj", "{C4043ABB-EC40-4194-B15B-C0D13C2CF5C8}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Tx.Bond", "..\..\Source\Tx.Bond\Tx.Bond.csproj", "{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug40|Any CPU = Debug40|Any CPU
@@ -39,6 +43,14 @@ Global
 		{C4043ABB-EC40-4194-B15B-C0D13C2CF5C8}.Release40|Any CPU.Build.0 = Release40|Any CPU
 		{C4043ABB-EC40-4194-B15B-C0D13C2CF5C8}.Release45|Any CPU.ActiveCfg = Release45|Any CPU
 		{C4043ABB-EC40-4194-B15B-C0D13C2CF5C8}.Release45|Any CPU.Build.0 = Release45|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Debug40|Any CPU.ActiveCfg = Debug40|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Debug40|Any CPU.Build.0 = Debug40|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Debug45|Any CPU.ActiveCfg = Debug45|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Debug45|Any CPU.Build.0 = Debug45|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Release40|Any CPU.ActiveCfg = Release40|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Release40|Any CPU.Build.0 = Release40|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Release45|Any CPU.ActiveCfg = Release45|Any CPU
+		{6DCE4A40-4946-41B0-ABE5-CE1700B598E9}.Release45|Any CPU.Build.0 = Release45|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE