{"id":64,"date":"2010-04-12T21:16:13","date_gmt":"2010-04-13T03:16:13","guid":{"rendered":"https:\/\/www.chrisedwards.dreamhosters.com\/blog\/?p=64"},"modified":"2010-06-14T23:04:20","modified_gmt":"2010-06-15T05:04:20","slug":"a-streaming-message-writer-reader-in-c-json-net","status":"publish","type":"post","link":"http:\/\/architester.com\/blog\/2010\/04\/12\/a-streaming-message-writer-reader-in-c-json-net\/","title":{"rendered":"A Streaming Message Writer & Reader in C# & Json.NET"},"content":{"rendered":"

Here is a C# implementation of a high performance message reader and writer that can read and write messages to any stream using Json.NET<\/a>. We are using a similar implementation here at BancVue<\/a> as our message store and it is performing quite well.<\/p>\n

I settled on Json.NET<\/a> after trying several other serializers and reading a<\/a> few<\/a> posts<\/a> on serialization performance. Objects serialized as Json are much smaller than Xml, and the Json.NET project seems to have the fastest serializer, and pretty wide support in the developer community.<\/p>\n

Overview<\/h3>\n

What I set out to create was something that can serialize millions of messages to a temporary holding place, then deserialize them for processing later. This serializer simply writes the Json to a file, but it could be used to write to any stream. There is a MessageWriter and a MessageReader. Together they can be used to form a \u00e2\u20ac\u0153Message Store\u00e2\u20ac\u009d as mentioned in some of Greg Young\u00e2\u20ac\u2122s<\/a> posts. Below is a unit test modeling how I want to use these objects.<\/p>\n

[ DataContract ]\r\ninternal class TestMessage\r\n{\r\n\t[ DataMember ]\r\n\tpublic string Text { get; set; }\r\n}\r\n\r\npublic class Given_a_message_writer_and_reader_tied_to_the_same_stream : ContextSpecification\r\n{\r\n\tprotected MemoryStream _stream;\r\n\tprotected IMessageWriter _writer;\r\n\tprotected IMessageReader _reader;\r\n\r\n\tprotected override void SharedContext()\r\n\t{\r\n\t\t_stream = new MemoryStream();\r\n\t\t_writer = new MessageWriter( _stream );\r\n\t\t_reader = new MessageReader( _stream );\r\n\t}\r\n}\r\n\r\n[ Concern( typeof ( MessageReader ) ) ]\r\npublic class When_a_message_is_written_to_the_stream : Given_a_message_writer_and_reader_tied_to_the_same_stream\r\n{\r\n\tprivate TestMessage _message;\r\n\r\n\tprotected override void Context()\r\n\t{\r\n\t\t_message = new TestMessage {Text = \"TestValue\"};\r\n\t}\r\n\r\n\tprotected override void Because()\r\n\t{\r\n\t\t_writer.WriteMessage( _message );\r\n\t\t_writer.Flush();\r\n\r\n\t\t\/\/ Need to rewind to beginning of stream so we can read.\r\n\t\t_stream.Seek( 0, SeekOrigin.Begin );\r\n\t}\r\n\r\n\t[ Observation ]\r\n\tpublic void Should_be_able_to_retrieve_that_message_from_the_reader()\r\n\t{\r\n\t\tvar actualMessage = _reader.ReadMessage< TestMessage >();\r\n\t\tAssert.That( ( actualMessage ).Text, Is.EqualTo( _message.Text ) );\r\n\t}\r\n}<\/pre>\n
Simply calling writer.WriteMessage() will write the message to the stream. Calling reader.ReadMessage() is all that is needed to read a message from the stream. I added Flush() to the writer so that when writing to files I can use a buffer. Calling flush simply calls flush on the underlying stream.<\/p>\n
The Interfaces<\/h3>\n
Here are the interfaces for the writer and reader.<\/p>\n
public interface IMessageWriter : IDisposable\r\n{\r\n\tvoid WriteMessage( object message );\r\n\tvoid Close();\r\n\tvoid Flush();\r\n}\r\n\r\npublic interface IMessageReader : IDisposable\r\n{\r\n\tT ReadMessage< T >();\r\n\tbool Eof { get; }\r\n\tvoid Close();\r\n}<\/pre>\n
The Implementation<\/h3>\n
Now lets look at the implementations. First, the writer\u00e2\u20ac\u00a6<\/p>\n
public class MessageWriter : IMessageWriter\r\n{\r\n\tprivate readonly StreamWriter _writer;\r\n\tprivate readonly JsonSerializerSettings _jsonSerializerSettings;\r\n\r\n\tpublic MessageWriter( Stream stream )\r\n\t{\r\n\t\t_writer = new StreamWriter( stream );\r\n\r\n\t\t_jsonSerializerSettings =\r\n\t\t\tnew JsonSerializerSettings\r\n\t\t\t\t{\r\n\t\t\t\t\tDefaultValueHandling = DefaultValueHandling.Ignore,\r\n\t\t\t\t\tNullValueHandling = NullValueHandling.Ignore,\r\n\t\t\t\t\tMissingMemberHandling = MissingMemberHandling.Ignore,\r\n\t\t\t\t\tTypeNameHandling = TypeNameHandling.Objects\r\n\t\t\t\t};\r\n\t}\r\n\r\n\tpublic void WriteMessage( object message )\r\n\t{\r\n\t\t_writer.WriteLine( JsonConvert.SerializeObject( message,\r\n\t\t                                                Formatting.None,\r\n\t\t                                                _jsonSerializerSettings ) );\r\n\t}\r\n\r\n\tpublic void Close()\r\n\t{\r\n\t\t_writer.Close();\r\n\t}\r\n\r\n\tpublic void Flush()\r\n\t{\r\n\t\t_writer.Flush();\r\n\t}\r\n\r\n\tpublic void Dispose()\r\n\t{\r\n\t\tClose();\r\n\t}\r\n}<\/pre>\n
Something to note\u00e2\u20ac\u00a6 I am using WriteLine() to write each object. This separates them with a CRLF. I had to do this because Json.NET currently requires you to have only one root element\u00e2\u20ac\u00a6unless you want to deserialize all the root elements as an array. Since I have millions of messages, I can\u00e2\u20ac\u2122t create an array of that size in memory or I will run out of RAM. Until this is fixed in Json.NET, I will just use the WriteLine mechanism. It has worked well so far.<\/p>\n
As you may have guessed, I am using ReadLine() in the reader to retrieve each message. Here is the reader\u00e2\u20ac\u2122s implementation\u00e2\u20ac\u00a6<\/p>\n
public class MessageReader : IMessageReader\r\n{\r\n\tprivate readonly StreamReader _reader;\r\n\tprivate readonly JsonSerializer _serializer;\r\n\r\n\tpublic MessageReader( Stream stream )\r\n\t{\r\n\t\t_reader = new StreamReader( stream );\r\n\r\n\t\t_serializer = new JsonSerializer\r\n\t\t\t\t{\r\n\t\t\t\t\tDefaultValueHandling = DefaultValueHandling.Ignore,\r\n\t\t\t\t\tNullValueHandling = NullValueHandling.Ignore,\r\n\t\t\t\t\tMissingMemberHandling = MissingMemberHandling.Ignore,\r\n\t\t\t\t\tTypeNameHandling = TypeNameHandling.Objects\r\n\t\t\t\t};\r\n\t}\r\n\r\n\tpublic T ReadMessage< T >()\r\n\t{\r\n\t\treturn (T)ReadMessage();\r\n\t}\r\n\r\n\tpublic object ReadMessage()\r\n\t{\r\n\t\tstring line = _reader.ReadLine();\r\n\r\n\t\treturn _serializer.Deserialize(\r\n\t\t\t\tnew JsonTextReader( new StringReader( line ) ) );\r\n\t}\r\n\r\n\tpublic bool Eof\r\n\t{\r\n\t\tget { return _reader.EndOfStream; }\r\n\t}\r\n\r\n\tpublic void Close()\r\n\t{\r\n\t\t_reader.Close();\r\n\t}\r\n\r\n\tpublic void Dispose()\r\n\t{\r\n\t\tClose();\r\n\t}\r\n}<\/pre>\n
Getting Json.NET to deserialize your types<\/h3>\n
Notice the JsonSerializerSettings. Most of these are to compress the Json by not showing empty or null values, but notice the last setting:<\/p>\n
TypeNameHandling = TypeNameHandling.Objects<\/p>\n
This setting tells the serializer to insert the type names of your objects into the Json itself. It then can use it when deserializing to determine what type to create. It does increase the size of your Json, but it makes it much easier to work with since you don\u00e2\u20ac\u2122t have to know exactly what type of message you are deserializing.<\/p>\n
Closing<\/h3>\n
All in all, this serializer\/deserializer set is very fast. We are using a similar implementation here at BancVue<\/a> in our production environment and are very pleased with the results.<\/p>\n
FYI: Multiple message test<\/h3>\n
Here is a test that shows a bit more about how we use the serializer when working with multiple messages.<\/p>\n
[ Concern( typeof ( MessageReader ) ) ]\r\npublic class When_multiple_messages_are_written_to_the_stream : Given_a_message_writer_and_reader_tied_to_the_same_stream\r\n{\r\n\tprivate TestMessage _message1;\r\n\tprivate TestMessage _message2;\r\n\r\n\tprotected override void Context()\r\n\t{\r\n\t\t_message1 = new TestMessage {Text = \"TestValue1\"};\r\n\t\t_message2 = new TestMessage {Text = \"TestValue2\"};\r\n\t}\r\n\r\n\tprotected override void Because()\r\n\t{\r\n\t\t_writer.WriteMessage( _message1 );\r\n\t\t_writer.WriteMessage( _message2 );\r\n\t\t_writer.Flush();\r\n\r\n\t\t\/\/ Need to rewind to beginning of stream to read.\r\n\t\t_stream.Seek( 0, SeekOrigin.Begin );\r\n\t}\r\n\r\n\t[ Observation ]\r\n\tpublic void Should_be_able_to_retrieve_that_message_from_the_reader()\r\n\t{\r\n\t\tvar actualMessage1 = _reader.ReadMessage< TestMessage >();\r\n\t\tAssert.That( actualMessage1.Text, Is.EqualTo( _message1.Text ) );\r\n\r\n\t\tvar actualMessage2 = _reader.ReadMessage< TestMessage >();\r\n\t\tAssert.That( actualMessage2.Text, Is.EqualTo( _message2.Text ) );\r\n\t}\r\n}<\/pre>\n
-Chris<\/p>\n","protected":false},"excerpt":{"rendered":"
Here is a C# implementation of a high performance message reader and writer that can read and write messages to any stream using Json.NET. We are using a similar implementation here at BancVue as our message store and it is performing quite well. I settled on Json.NET after trying several other serializers and reading a […]<\/p>\n","protected":false},"author":2,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[27,31,20,21],"tags":[79,30,87,88],"class_list":["post-64","post","type-post","status-publish","format-standard","hentry","category-net","category-c","category-messaging","category-soa","tag-net","tag-csharp","tag-messaging","tag-soa"],"aioseo_notices":[],"_links":{"self":[{"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/posts\/64","targetHints":{"allow":["GET"]}}],"collection":[{"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/users\/2"}],"replies":[{"embeddable":true,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/comments?post=64"}],"version-history":[{"count":15,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/posts\/64\/revisions"}],"predecessor-version":[{"id":171,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/posts\/64\/revisions\/171"}],"wp:attachment":[{"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/media?parent=64"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/categories?post=64"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/architester.com\/blog\/wp-json\/wp\/v2\/tags?post=64"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}