diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..1ff0c42
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1,63 @@
+###############################################################################
+# Set default behavior to automatically normalize line endings.
+###############################################################################
+* text=auto
+
+###############################################################################
+# Set default behavior for command prompt diff.
+#
+# This is need for earlier builds of msysgit that does not have it on by
+# default for csharp files.
+# Note: This is only used by command line
+###############################################################################
+#*.cs     diff=csharp
+
+###############################################################################
+# Set the merge driver for project and solution files
+#
+# Merging from the command prompt will add diff markers to the files if there
+# are conflicts (Merging from VS is not affected by the settings below, in VS
+# the diff markers are never inserted). Diff markers may cause the following 
+# file extensions to fail to load in VS. An alternative would be to treat
+# these files as binary and thus will always conflict and require user
+# intervention with every merge. To do so, just uncomment the entries below
+###############################################################################
+#*.sln       merge=binary
+#*.csproj    merge=binary
+#*.vbproj    merge=binary
+#*.vcxproj   merge=binary
+#*.vcproj    merge=binary
+#*.dbproj    merge=binary
+#*.fsproj    merge=binary
+#*.lsproj    merge=binary
+#*.wixproj   merge=binary
+#*.modelproj merge=binary
+#*.sqlproj   merge=binary
+#*.wwaproj   merge=binary
+
+###############################################################################
+# behavior for image files
+#
+# image files are treated as binary by default.
+###############################################################################
+#*.jpg   binary
+#*.png   binary
+#*.gif   binary
+
+###############################################################################
+# diff behavior for common document formats
+# 
+# Convert binary document formats to text before diffing them. This feature
+# is only available from the command line. Turn it on by uncommenting the 
+# entries below.
+###############################################################################
+#*.doc   diff=astextplain
+#*.DOC   diff=astextplain
+#*.docx  diff=astextplain
+#*.DOCX  diff=astextplain
+#*.dot   diff=astextplain
+#*.DOT   diff=astextplain
+#*.pdf   diff=astextplain
+#*.PDF   diff=astextplain
+#*.rtf   diff=astextplain
+#*.RTF   diff=astextplain
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..6e2e10d
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,261 @@
+## Ignore Visual Studio temporary files, build results, and
+## files generated by popular Visual Studio add-ons.
+
+# User-specific files
+*.suo
+*.user
+*.userosscache
+*.sln.docstates
+
+# User-specific files (MonoDevelop/Xamarin Studio)
+*.userprefs
+
+# Build results
+[Dd]ebug/
+[Dd]ebugPublic/
+[Rr]elease/
+[Rr]eleases/
+x64/
+x86/
+bld/
+[Bb]in/
+[Oo]bj/
+[Ll]og/
+
+# Visual Studio 2015 cache/options directory
+.vs/
+# Uncomment if you have tasks that create the project's static files in wwwroot
+wwwroot/
+
+# MSTest test Results
+[Tt]est[Rr]esult*/
+[Bb]uild[Ll]og.*
+
+# NUNIT
+*.VisualState.xml
+TestResult.xml
+
+# Build Results of an ATL Project
+[Dd]ebugPS/
+[Rr]eleasePS/
+dlldata.c
+
+# DNX
+project.lock.json
+project.fragment.lock.json
+artifacts/
+
+*_i.c
+*_p.c
+*_i.h
+*.ilk
+*.meta
+*.obj
+*.pch
+*.pdb
+*.pgc
+*.pgd
+*.rsp
+*.sbr
+*.tlb
+*.tli
+*.tlh
+*.tmp
+*.tmp_proj
+*.log
+*.vspscc
+*.vssscc
+.builds
+*.pidb
+*.svclog
+*.scc
+
+# Chutzpah Test files
+_Chutzpah*
+
+# Visual C++ cache files
+ipch/
+*.aps
+*.ncb
+*.opendb
+*.opensdf
+*.sdf
+*.cachefile
+*.VC.db
+*.VC.VC.opendb
+
+# Visual Studio profiler
+*.psess
+*.vsp
+*.vspx
+*.sap
+
+# TFS 2012 Local Workspace
+$tf/
+
+# Guidance Automation Toolkit
+*.gpState
+
+# ReSharper is a .NET coding add-in
+_ReSharper*/
+*.[Rr]e[Ss]harper
+*.DotSettings.user
+
+# JustCode is a .NET coding add-in
+.JustCode
+
+# TeamCity is a build add-in
+_TeamCity*
+
+# DotCover is a Code Coverage Tool
+*.dotCover
+
+# NCrunch
+_NCrunch_*
+.*crunch*.local.xml
+nCrunchTemp_*
+
+# MightyMoose
+*.mm.*
+AutoTest.Net/
+
+# Web workbench (sass)
+.sass-cache/
+
+# Installshield output folder
+[Ee]xpress/
+
+# DocProject is a documentation generator add-in
+DocProject/buildhelp/
+DocProject/Help/*.HxT
+DocProject/Help/*.HxC
+DocProject/Help/*.hhc
+DocProject/Help/*.hhk
+DocProject/Help/*.hhp
+DocProject/Help/Html2
+DocProject/Help/html
+
+# Click-Once directory
+publish/
+
+# Publish Web Output
+*.[Pp]ublish.xml
+*.azurePubxml
+# TODO: Comment the next line if you want to checkin your web deploy settings
+# but database connection strings (with potential passwords) will be unencrypted
+#*.pubxml
+*.publishproj
+
+# Microsoft Azure Web App publish settings. Comment the next line if you want to
+# checkin your Azure Web App publish settings, but sensitive information contained
+# in these scripts will be unencrypted
+PublishScripts/
+
+# NuGet Packages
+*.nupkg
+# The packages folder can be ignored because of Package Restore
+**/packages/*
+# except build/, which is used as an MSBuild target.
+!**/packages/build/
+# Uncomment if necessary however generally it will be regenerated when needed
+#!**/packages/repositories.config
+# NuGet v3's project.json files produces more ignoreable files
+*.nuget.props
+*.nuget.targets
+
+# Microsoft Azure Build Output
+csx/
+*.build.csdef
+
+# Microsoft Azure Emulator
+ecf/
+rcf/
+
+# Windows Store app package directories and files
+AppPackages/
+BundleArtifacts/
+Package.StoreAssociation.xml
+_pkginfo.txt
+
+# Visual Studio cache files
+# files ending in .cache can be ignored
+*.[Cc]ache
+# but keep track of directories ending in .cache
+!*.[Cc]ache/
+
+# Others
+ClientBin/
+~$*
+*~
+*.dbmdl
+*.dbproj.schemaview
+*.jfm
+*.pfx
+*.publishsettings
+node_modules/
+orleans.codegen.cs
+
+# Since there are multiple workflows, uncomment next line to ignore bower_components
+# (https://github.com/github/gitignore/pull/1529#issuecomment-104372622)
+#bower_components/
+
+# RIA/Silverlight projects
+Generated_Code/
+
+# Backup & report files from converting an old project file
+# to a newer Visual Studio version. Backup files are not needed,
+# because we have git ;-)
+_UpgradeReport_Files/
+Backup*/
+UpgradeLog*.XML
+UpgradeLog*.htm
+
+# SQL Server files
+*.mdf
+*.ldf
+
+# Business Intelligence projects
+*.rdl.data
+*.bim.layout
+*.bim_*.settings
+
+# Microsoft Fakes
+FakesAssemblies/
+
+# GhostDoc plugin setting file
+*.GhostDoc.xml
+
+# Node.js Tools for Visual Studio
+.ntvs_analysis.dat
+
+# Visual Studio 6 build log
+*.plg
+
+# Visual Studio 6 workspace options file
+*.opt
+
+# Visual Studio LightSwitch build output
+**/*.HTMLClient/GeneratedArtifacts
+**/*.DesktopClient/GeneratedArtifacts
+**/*.DesktopClient/ModelManifest.xml
+**/*.Server/GeneratedArtifacts
+**/*.Server/ModelManifest.xml
+_Pvt_Extensions
+
+# Paket dependency manager
+.paket/paket.exe
+paket-files/
+
+# FAKE - F# Make
+.fake/
+
+# JetBrains Rider
+.idea/
+*.sln.iml
+
+# CodeRush
+.cr/
+
+# Python Tools for Visual Studio (PTVS)
+__pycache__/
+*.pyc
\ No newline at end of file
diff --git a/LICENSE b/LICENSE
new file mode 100644
index 0000000..b3ae3d2
--- /dev/null
+++ b/LICENSE
@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright 2018 Daniel Blankensteiner
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..4bb5697
--- /dev/null
+++ b/README.md
@@ -0,0 +1,216 @@
+# DotPulsar
+
+Native .NET/C# client library for [Apache Pulsar](https://pulsar.apache.org/)
+
+## Getting Started
+
+DotPulsar is written entirely in C# and implement Apache Pulsar's [binary protocol](https://pulsar.apache.org/docs/en/develop-binary-protocol/). Other options was using the [C++ client library](https://pulsar.apache.org/docs/en/client-libraries-cpp/) (which is what the [Python client](https://pulsar.apache.org/docs/en/client-libraries-python/) and [Go client](https://pulsar.apache.org/docs/en/client-libraries-go/) do) or build on top of the [WebSocket API](https://pulsar.apache.org/docs/en/client-libraries-websocket/). We decided to implement the binary protocol in order to gain full control and maximize portability and performance.
+
+DotPulsar's API is strongly inspired by Apache Pulsar's official [Java client](https://pulsar.apache.org/docs/en/client-libraries-java/), but a 100% match is not goal.
+
+Let's see how to produce, consume and read messages.
+
+### Producing messages
+
+Producers can be created via the extension method show below, which follows the API from the Java client:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var producer = client.NewProducer().Topic("persistent://public/default/mytopic").Create();
+await producer.Send(System.Text.Encoding.UTF8.GetBytes("Hello World"));
+```
+
+If you are not a fan of extensions methods and builders, then there is another option:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var producerOptions = new ProducerOptions
+{
+    ProducerName = "MyProducer",
+    Topic = "persistent://public/default/mytopic"
+};
+var producer = client.CreateProducer(producerOptions);
+```
+
+In the above you only specify the data being sent, but you can also specify metadata:
+
+```csharp
+var data = Encoding.UTF8.GetBytes("Hello World");
+var messageId = await producer.NewMessage()
+                              .Property("SomeKey", "SomeValue") //EventTime and SequenceId can also be set
+                              .Send(data);
+```
+
+If you are not a fan of extensions methods and builders, then there is another option:
+
+```csharp
+var data = Encoding.UTF8.GetBytes("Hello World");
+var metadata = new MessageMetadata(); //EventTime and SequenceId can be set via properties
+metadata["SomeKey"] = "SomeValue";
+var messageId = await producer.Send(metadata, data));
+```
+
+### Consuming messages
+
+Consumers can be created via the extension method show below, which follows the API from the Java client:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var consumer = client.NewConsumer()
+                     .SubscriptionName("MySubscription")
+                     .Topic("persistent://public/default/mytopic")
+                     .Create();
+var message = await consumer.Receive();
+Console.WriteLine("Received Message: " + Encoding.UTF8.GetString(message.Data.ToArray()));
+await consumer.Acknowledge(message);
+```
+
+If you are not a fan of extensions methods and builders, then there is another option:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var consumerOptions = new ConsumerOptions
+{
+    SubscriptionName = "MySubscription",
+    Topic = "persistent://public/default/mytopic"
+};
+var consumer = client.CreateConsumer(consumerOptions);
+```
+
+### Reading messages
+
+Readers can be created via the extension method show below, which follows the API from the Java client:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var reader = client.NewReader()
+                   .StartMessageId(MessageId.Earliest)
+                   .Topic("persistent://public/default/mytopic")
+                   .Create();
+var message = await reader.Receive();
+Console.WriteLine("Received Message: " + Encoding.UTF8.GetString(message.Data.ToArray()));
+```
+
+If you are not a fan of extensions methods and builders, then there is another option:
+
+```csharp
+var client = PulsarClient.Builder().Build();
+var readerOptions = new ReaderOptions
+{
+    StartMessageId = MessageId.Earliest,
+    Topic = "persistent://public/default/mytopic"
+};
+var reader = client.CreateReader(readerOptions);
+```
+
+## Monitoring state
+
+Consumers, producers and readers all have states that can be monitored. Let's have a look at what states they can have.
+
+### Consumer states
+
+* Active (All is well)
+* Inactive (All is well. The subscription type is 'Failover' and you are not the active consumer)
+* Closed (The consumer or PulsarClient has been disposed)
+* Disconnected (The connection was lost and attempts are being made to reconnect)
+* Faulted (An unrecoverable error has occurred)
+* ReachedEndOfTopic (No more messages will be delivered)
+
+### Producer states
+
+* Closed (The producer or PulsarClient has been disposed)
+* Connected (All is well)
+* Disconnected (The connection was lost and attempts are being made to reconnect)
+* Faulted (An unrecoverable error has occurred)
+
+### Reader states
+
+* Closed (The reader or PulsarClient has been disposed)
+* Connected: (All is well)
+* Disconnected (The connection was lost and attempts are being made to reconnect)
+* Faulted (An unrecoverable error has occurred)
+* ReachedEndOfTopic (No more messages will be delivered)
+
+### How to
+
+Monitoring the state is easy, so let's see how to monitor when a consumer changes state:
+
+```csharp
+private static async Task MonitorConsumerState(IConsumer consumer, CancellationToken cancellationToken)
+{
+	var state = ConsumerState.Disconnected;
+
+	while (true)
+	{
+		state = await consumer.StateChangedFrom(state, cancellationToken);
+
+		switch (state)
+		{
+			case ConsumerState.Active:
+				Console.WriteLine("Consumer is active");
+				break;
+			case ConsumerState.Inactive:
+				Console.WriteLine("Consumer is inactive");
+				break;
+			case ConsumerState.Disconnected:
+				Console.WriteLine("Consumer is disconnected");
+				break;
+			case ConsumerState.Closed:
+				Console.WriteLine("Consumer has closed");
+				break;
+			case ConsumerState.ReachedEndOfTopic:
+				Console.WriteLine("Consumer has reached end of topic");
+				break;
+			case ConsumerState.Faulted:
+				Console.WriteLine("Consumer has faulted");
+				break;
+		}
+
+		if (consumer.IsFinalState(state))
+			return;
+	}
+}
+```
+
+Here the variable 'state' will contained to new state. You can both monitor going From (StateChangedFrom) and To (StateChangedTo) a state. 
+Some states are final, meaning the state can no longer change. For consumers 'Closed', 'Faulted' and 'ReachedEndOfTopic' are final states. When the consumer enter a final state, all monitoring tasks are completed. So if you e.g. are monitoring going to 'Diconnected' and the consumer is 'Closed', then you task will complete and return 'Closed'.
+
+## Built With
+
+* [protobuf-net](https://github.com/mgravell/protobuf-net) - Provides simple access to fast and efficient "Protocol Buffers" serialization from .NET applications
+* [System.IO.Pipelines](https://www.nuget.org/packages/System.IO.Pipelines/) - Single producer single consumer byte buffer management
+
+## Versioning
+
+We use [SemVer](http://semver.org/) for versioning. For the versions available, see the [tags on this repository](https://github.com/danske-commodities/dotpulsar/tags).
+
+## Authors
+
+* **Daniel Blankensteiner** - *Initial work* - [Danske Commodities](https://github.com/danske-commodities)
+
+See also the list of [contributors](https://github.com/danske-commodities/dotpulsar/contributors) who participated in this project.
+
+## License
+
+This project is licensed under the Apache License Version 2.0 - see the [LICENSE](LICENSE) file for details.
+
+## Roadmap
+
+1.0.0
+
+* Move to IAsyncDisposable and IAsyncEnumerable (will mean moving to .NET Standard 2.1)
+
+X.X.X //Future
+
+* Schemas
+* Authentication/Authorization (TLS Authentication, Athenz, Kerberos)
+* Partitioned topics
+* Topic compaction
+* Message compression (LZ4, ZLIB, ZSTD, SNAPPY)
+* Multi-topic subscriptions
+* Connection encryption
+* Message encryption
+* Batching
+* CommandConsumerStats/CommandConsumerStatsResponse
+* CommandGetTopicsOfNamespace/CommandGetTopicsOfNamespaceResponse
+* CommandPartitionedTopicMetadata/CommandPartitionedTopicMetadataResponse
diff --git a/src/Directory.Build.props b/src/Directory.Build.props
new file mode 100644
index 0000000..13d206a
--- /dev/null
+++ b/src/Directory.Build.props
@@ -0,0 +1,7 @@
+<Project>
+
+  <PropertyGroup>
+    <LangVersion>8.0</LangVersion>
+  </PropertyGroup>
+
+</Project>
\ No newline at end of file
diff --git a/src/DotPulsar.Tests/DotPulsar.Tests.csproj b/src/DotPulsar.Tests/DotPulsar.Tests.csproj
new file mode 100644
index 0000000..34b3de6
--- /dev/null
+++ b/src/DotPulsar.Tests/DotPulsar.Tests.csproj
@@ -0,0 +1,21 @@
+﻿<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <IsPackable>false</IsPackable>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.2.0" />
+    <PackageReference Include="xunit" Version="2.4.1" />
+    <PackageReference Include="xunit.runner.visualstudio" Version="2.4.1">
+      <PrivateAssets>all</PrivateAssets>
+      <IncludeAssets>runtime; build; native; contentfiles; analyzers</IncludeAssets>
+    </PackageReference>
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\DotPulsar\DotPulsar.csproj" />
+  </ItemGroup>
+
+</Project>
diff --git a/src/DotPulsar.Tests/Internal/AsyncLockTests.cs b/src/DotPulsar.Tests/Internal/AsyncLockTests.cs
new file mode 100644
index 0000000..b5ba9b1
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/AsyncLockTests.cs
@@ -0,0 +1,137 @@
+﻿using DotPulsar.Internal;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class AsyncLockTests
+    {
+        [Fact]
+        public void Lock_GivenLockIsFree_ShouldReturnCompletedTask()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+
+            //Act
+            var actual = sut.Lock();
+
+            //Assert
+            Assert.True(actual.IsCompleted);
+
+            //Annihilate 
+            actual.Result.Dispose();
+            sut.Dispose();
+        }
+
+        [Fact]
+        public async Task Lock_GivenLockIsTaken_ShouldReturnIncompletedTask()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+            var alreadyTaken = await sut.Lock();
+
+            //Act
+            var actual = sut.Lock();
+
+            //Assert
+            Assert.False(actual.IsCompleted);
+
+            //Annihilate
+            alreadyTaken.Dispose();
+            actual.Result.Dispose();
+            sut.Dispose();
+        }
+
+        [Fact]
+        public async Task Lock_GivenLockIsDisposed_ShouldThrowObjectDisposedException()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+            sut.Dispose();
+
+            //Act
+            var exception = await Record.ExceptionAsync(() => sut.Lock());
+
+            //Assert
+            Assert.IsType<ObjectDisposedException>(exception);
+        }
+
+        [Fact]
+        public async Task Lock_GivenLockIsDisposedWhileAwaitingLock_ShouldThrowObjectDisposedException()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+            var gotLock = await sut.Lock();
+            var awaiting = sut.Lock();
+            _ = Task.Run(() => sut.Dispose());
+
+            //Act
+            var exception = await Record.ExceptionAsync(() => awaiting);
+
+            //Assert
+            Assert.IsType<ObjectDisposedException>(exception);
+
+            //Annihilate
+            sut.Dispose();
+            gotLock.Dispose();
+        }
+
+        [Fact]
+        public async Task Lock_GivenLockIsTakenAndCancellationTokenIsActivated_ShouldThrowTaskCanceledException()
+        {
+            //Arrange
+            var cts = new CancellationTokenSource();
+            var sut = new AsyncLock();
+            var gotLock = await sut.Lock();
+            var awaiting = sut.Lock(cts.Token);
+
+            //Act
+            cts.Cancel();
+            var exception = await Record.ExceptionAsync(() => awaiting);
+
+            //Assert
+            Assert.IsType<TaskCanceledException>(exception);
+
+            //Annihilate
+            cts.Dispose();
+            gotLock.Dispose();
+            sut.Dispose();
+        }
+
+        [Fact]
+        public async Task Dispose_GivenLockIsDisposedWhileItIsTaken_ShouldNotCompleteBeforeItIsReleased()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+            var gotLock = await sut.Lock();
+            var disposeTask = Task.Run(() => sut.Dispose());
+            Assert.False(disposeTask.IsCompleted);
+
+            //Act
+            gotLock.Dispose();
+            await disposeTask;
+
+            //Assert
+            Assert.True(disposeTask.IsCompleted);
+
+            //Annihilate
+            sut.Dispose();
+        }
+
+        [Fact]
+        public void Dispose_WhenCalledMultipleTimes_ShouldBeSafeToDoSo()
+        {
+            //Arrange
+            var sut = new AsyncLock();
+
+            //Act
+            sut.Dispose();
+            var exception = Record.Exception(() => sut.Dispose());
+
+            //Assert
+            Assert.Null(exception);
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/AsyncQueueTests.cs b/src/DotPulsar.Tests/Internal/AsyncQueueTests.cs
new file mode 100644
index 0000000..4491804
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/AsyncQueueTests.cs
@@ -0,0 +1,117 @@
+using DotPulsar.Internal;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class AsyncQueueTests
+    {
+        [Fact]
+        public async Task Enqueue_GivenDequeueTaskWasWaiting_ShouldCompleteDequeueTask()
+        {
+            //Arrange
+            const int value = 1;
+            var queue = new AsyncQueue<int>();
+            var dequeueTask = queue.Dequeue();
+            queue.Enqueue(value);
+
+            //Act
+            var actual = await dequeueTask;
+
+            //Assert
+            Assert.Equal(value, actual);
+
+            //Annihilate
+            queue.Dispose();
+        }
+
+        [Fact]
+        public async Task DequeueAsync_GivenQueueWasNotEmpty_ShouldCompleteDequeueTask()
+        {
+            //Arrange
+            const int value = 1;
+            var queue = new AsyncQueue<int>();
+            queue.Enqueue(value);
+
+            //Act
+            var actual = await queue.Dequeue();
+
+            //Assert
+            Assert.Equal(value, actual);
+
+            //Annihilate
+            queue.Dispose();
+        }
+
+        [Fact]
+        public async Task DequeueAsync_GivenMultipleDequeues_ShouldCompleteInOrderedSequence()
+        {
+            //Arrange
+            const int value1 = 1, value2 = 2;
+            var queue = new AsyncQueue<int>();
+            var dequeue1 = queue.Dequeue();
+            var dequeue2 = queue.Dequeue();
+            queue.Enqueue(value1);
+            queue.Enqueue(value2);
+
+            //Act
+            var actual1 = await dequeue1;
+            var actual2 = await dequeue2;
+
+            //Assert
+            Assert.Equal(value1, actual1);
+            Assert.Equal(value2, actual2);
+
+            //Annihilate
+            queue.Dispose();
+        }
+
+        [Fact]
+        public async Task DequeueAsync_GivenSequenceOfInput_ShouldReturnSameSequenceOfOutput()
+        {
+            //Arrange
+            const int value1 = 1, value2 = 2;
+            var queue = new AsyncQueue<int>();
+            queue.Enqueue(value1);
+            queue.Enqueue(value2);
+
+            //Act
+            var actual1 = await queue.Dequeue();
+            var actual2 = await queue.Dequeue();
+
+            //Assert
+            Assert.Equal(value1, actual1);
+            Assert.Equal(value2, actual2);
+
+            //Annihilate
+            queue.Dispose();
+        }
+
+        [Fact]
+        public async Task DequeueAsync_GivenTokenIsCanceled_ShouldCancelTask()
+        {
+            //Arrange
+            CancellationTokenSource source1 = new CancellationTokenSource(), source2 = new CancellationTokenSource();
+            const int excepted = 1;
+            var queue = new AsyncQueue<int>();
+            var task1 = queue.Dequeue(source1.Token);
+            var task2 = queue.Dequeue(source2.Token);
+
+            //Act
+            source1.Cancel();
+            queue.Enqueue(excepted);
+            var exception = await Record.ExceptionAsync(() => task1);
+            await task2;
+
+            //Assert
+            Assert.IsType<TaskCanceledException>(exception);
+            Assert.Equal(excepted, task2.Result);
+
+            //Annihilate
+            source1.Dispose();
+            source2.Dispose();
+            queue.Dispose();
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/Crc32CTests.cs b/src/DotPulsar.Tests/Internal/Crc32CTests.cs
new file mode 100644
index 0000000..0187c29
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/Crc32CTests.cs
@@ -0,0 +1,39 @@
+﻿using DotPulsar.Internal;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class Crc32CTests
+    {
+        [Fact]
+        public void Calculate_GivenSequenceWithSingleSegment_ShouldReturnExpectedChecksum()
+        {
+            //Arrange
+            var segment = new byte[] { 0x10, 0x01, 0x18, 0xc9, 0xf8, 0x86, 0x94, 0xeb, 0x2c };
+            var sequence = new SequenceBuilder<byte>().Append(segment).Build();
+
+            //Act
+            var actual = Crc32C.Calculate(sequence);
+
+            //Assert
+            const uint expected = 2355953212;
+            Assert.Equal(expected, actual);
+        }
+
+        [Fact]
+        public void Calculate_GivenSequenceWithMultipleSegments_ShouldReturnExpectedChecksum()
+        {
+            //Arrange
+            var s1 = new byte[] { 0x0a, 0x0f, 0x73, 0x74, 0x61, 0x6e, 0x64, 0x61, 0x6c, 0x6f, 0x6e, 0x65, 0x2d, 0x33, 0x30, 0x2d, 0x35, 0x10, 0x00, 0x18, 0xc7, 0xee, 0xa3, 0x93, 0xeb, 0x2c, 0x58, 0x01 };
+            var s2 = new byte[] { 0x10, 0x01, 0x18, 0xc9, 0xf8, 0x86, 0x94, 0xeb, 0x2c };
+            var sequence = new SequenceBuilder<byte>().Append(s1).Append(s2).Build();
+
+            //Act
+            var actual = Crc32C.Calculate(sequence);
+
+            //Assert
+            const uint expected = 1079987866;
+            Assert.Equal(expected, actual);
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/Extensions/ReadOnlySequenceExtensionsTests.cs b/src/DotPulsar.Tests/Internal/Extensions/ReadOnlySequenceExtensionsTests.cs
new file mode 100644
index 0000000..82f92e8
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/Extensions/ReadOnlySequenceExtensionsTests.cs
@@ -0,0 +1,146 @@
+﻿using DotPulsar.Internal;
+using DotPulsar.Internal.Extensions;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal.Extensions
+{
+    public class ReadOnlySequenceExtensionsTests
+    {
+        [Fact]
+        public void StartsWith_GivenToShortSequenceWithSingleSegment_ShouldReturnFalse()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01 });
+
+            //Assert
+            Assert.False(actual);
+        }
+
+        [Fact]
+        public void StartsWith_GivenSequenceWithSingleSegment_ShouldReturnFalse()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x02, 0x01 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01 });
+
+            //Assert
+            Assert.False(actual);
+        }
+
+        [Fact]
+        public void StartsWith_GivenSequenceWithSingleSegment_ShouldReturnTrue()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x01, 0x02 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01 });
+
+            //Assert
+            Assert.True(actual);
+        }
+
+        [Fact]
+        public void StartsWith_GivenToShortSequenceWithMultipleSegments_ShouldReturnFalse()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x01 }).Append(new byte[] { 0x02 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01, 0x02, 0x03 });
+
+            //Assert
+            Assert.False(actual);
+        }
+
+        [Fact]
+        public void StartsWith_GivenSequenceWithMultipleSegments_ShouldReturnFalse()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x02 }).Append(new byte[] { 0x01, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01, 0x02 });
+
+            //Assert
+            Assert.False(actual);
+        }
+
+        [Fact]
+        public void StartsWith_GivenSequenceWithMultipleSegments_ShouldReturnTrue()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x01 }).Append(new byte[] { 0x02, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.StartsWith(new byte[] { 0x00, 0x01, 0x02 });
+
+            //Assert
+            Assert.True(actual);
+        }
+
+        [Fact]
+        public void ReadUInt32_GivenSequenceWithSingleSegment_ShouldGiveExceptedResult()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x01, 0x02, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.ReadUInt32(0, true);
+
+            //Assert
+            const uint expected = 66051;
+            Assert.Equal(expected, actual);
+        }
+
+        [Fact]
+        public void ReadUInt32_GivenSequenceWithSingleSegmentAndNonZeroStart_ShouldGiveExceptedResult()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x09, 0x00, 0x01, 0x02, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.ReadUInt32(1, true);
+
+            //Assert
+            const uint expected = 66051;
+            Assert.Equal(expected, actual);
+        }
+
+        [Fact]
+        public void ReadUInt32_GivenSequenceWithMultipleSegments_ShouldGiveExceptedResult()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>().Append(new byte[] { 0x00, 0x01 }).Append(new byte[] { 0x02, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.ReadUInt32(0, true);
+
+            //Assert
+            const uint expected = 66051;
+            Assert.Equal(expected, actual);
+        }
+
+        [Fact]
+        public void ReadUInt32_GivenSequenceWithMultipleSegmentsAndNonZeroStart_ShouldGiveExceptedResult()
+        {
+            //Arrange
+            var sequence = new SequenceBuilder<byte>()
+                .Append(new byte[] { 0x09, 0x09, 0x09 })
+                .Append(new byte[] { 0x09, 0x00, 0x01 })
+                .Append(new byte[] { 0x02, 0x03 }).Build();
+
+            //Act
+            var actual = sequence.ReadUInt32(4, true);
+
+            //Assert
+            const uint expected = 66051;
+            Assert.Equal(expected, actual);
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/SequenceBuilderTests.cs b/src/DotPulsar.Tests/Internal/SequenceBuilderTests.cs
new file mode 100644
index 0000000..51c0ab7
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/SequenceBuilderTests.cs
@@ -0,0 +1,45 @@
+﻿using DotPulsar.Internal;
+using System.Buffers;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class SequenceBuilderTests
+    {
+        [Fact]
+        public void Append_GivenMultipleInput_ShouldArrangeInCorrectOrder()
+        {
+            //Arrange
+            var a = new byte[] { 0x00, 0x01, 0x02, 0x03 };
+            var b = new byte[] { 0x04, 0x05, 0x06, 0x07, 0x08 };
+            var c = new byte[] { 0x09 };
+            var builder = new SequenceBuilder<byte>().Append(a).Append(b).Append(c);
+
+            //Act
+            var sequence = builder.Build();
+
+            //Assert
+            var array = sequence.ToArray();
+            for (byte i = 0; i < array.Length; ++i)
+                Assert.Equal(i, array[i]);
+        }
+
+        [Fact]
+        public void Prepend_GivenMultipleInput_ShouldArrangeInCorrectOrder()
+        {
+            //Arrange
+            var a = new byte[] { 0x00, 0x01, 0x02, 0x03 };
+            var b = new byte[] { 0x04, 0x05, 0x06, 0x07, 0x08 };
+            var c = new byte[] { 0x09 };
+            var builder = new SequenceBuilder<byte>().Prepend(c).Prepend(b).Prepend(a);
+
+            //Act
+            var sequence = builder.Build();
+
+            //Assert
+            var array = sequence.ToArray();
+            for (byte i = 0; i < array.Length; ++i)
+                Assert.Equal(i, array[i]);
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/SerializerTests.cs b/src/DotPulsar.Tests/Internal/SerializerTests.cs
new file mode 100644
index 0000000..f933a74
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/SerializerTests.cs
@@ -0,0 +1,22 @@
+﻿using DotPulsar.Internal;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class SerializerTests
+    {
+        [Fact]
+        public void ToBigEndianBytes_GivenUnsignedInteger_ShouldReturnExpectedBytes()
+        {
+            //Arrange
+            uint value = 66051;
+
+            //Act
+            var actual = Serializer.ToBigEndianBytes(value);
+
+            //Assert
+            var expected = new byte[] { 0x00, 0x01, 0x02, 0x03 };
+            Assert.Equal(expected, actual);
+        }
+    }
+}
diff --git a/src/DotPulsar.Tests/Internal/StateManagerTests.cs b/src/DotPulsar.Tests/Internal/StateManagerTests.cs
new file mode 100644
index 0000000..b7a6590
--- /dev/null
+++ b/src/DotPulsar.Tests/Internal/StateManagerTests.cs
@@ -0,0 +1,231 @@
+﻿using DotPulsar.Internal;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace DotPulsar.Tests.Internal
+{
+    public class StateManagerTests
+    {
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Connected, ProducerState.Connected)]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected, ProducerState.Connected)]
+        [InlineData(ProducerState.Connected, ProducerState.Closed, ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Disconnected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Closed, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Closed, ProducerState.Connected, ProducerState.Closed)]
+        [InlineData(ProducerState.Closed, ProducerState.Disconnected, ProducerState.Closed)]
+        [InlineData(ProducerState.Closed, ProducerState.Closed, ProducerState.Closed)]
+        public void SetState_GivenNewState_ShouldReturnFormerState(ProducerState initialState, ProducerState newState, ProducerState expected)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+
+            //Act
+            var actual = sut.SetState(newState);
+
+            //Assert
+            Assert.Equal(expected, actual);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected)]
+        [InlineData(ProducerState.Closed)]
+        public void SetState_GivenStateIsFinal_ShouldNotChangeState(ProducerState newState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(ProducerState.Closed, ProducerState.Closed);
+
+            //Act
+            _ = sut.SetState(newState);
+
+            //Assert
+            Assert.Equal(ProducerState.Closed, sut.CurrentState);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Connected, ProducerState.Closed)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Closed)]
+        public void SetState_GivenStateIsChangedToWanted_ShouldCompleteTask(ProducerState initialState, ProducerState newState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+            var task = sut.StateChangedTo(newState, default);
+
+            //Act
+            _ = sut.SetState(newState);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Connected, ProducerState.Closed)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Closed)]
+        public void SetState_GivenStateIsChangedFromWanted_ShouldCompleteTask(ProducerState initialState, ProducerState newState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+            var task = sut.StateChangedFrom(initialState, default);
+
+            //Act
+            _ = sut.SetState(newState);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected)]
+        [InlineData(ProducerState.Closed)]
+        public void StateChangedTo_GivenStateIsAlreadyWanted_ShouldCompleteTask(ProducerState state)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(state, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedTo(state, default);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Connected, ProducerState.Closed)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Closed)]
+        public void StateChangedTo_GivenStateIsNotWanted_ShouldNotCompleteTask(ProducerState initialState, ProducerState wantedState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedTo(wantedState, default);
+
+            //Assert
+            Assert.False(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected)]
+        public void StateChangedTo_GivenStateIsFinal_ShouldCompleteTask(ProducerState state)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(ProducerState.Closed, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedTo(state, default);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected)]
+        public void StateChangedFrom_GivenStateHasNotChanged_ShouldNotCompleteTask(ProducerState state)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(state, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedFrom(state, default);
+
+            //Assert
+            Assert.False(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Connected, ProducerState.Closed)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Closed)]
+        public void StateChangedFrom_GivenStateHasChanged_ShouldCompleteTask(ProducerState initialState, ProducerState fromState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedFrom(fromState, default);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected)]
+        [InlineData(ProducerState.Disconnected)]
+        [InlineData(ProducerState.Closed)]
+        public void StateChangedFrom_GivenStateIsFinal_ShouldCompleteTask(ProducerState state)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(ProducerState.Closed, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedFrom(state, default);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected)]
+        public void SetState_GivenStateIsChangeToFinalState_ShouldCompleteTask(ProducerState initialState, ProducerState wantedState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedTo(wantedState, default);
+            _ = sut.SetState(ProducerState.Closed);
+
+            //Assert
+            Assert.True(task.IsCompleted);
+        }
+
+        [Theory]
+        [InlineData(ProducerState.Connected, ProducerState.Disconnected, ProducerState.Closed)]
+        [InlineData(ProducerState.Disconnected, ProducerState.Connected, ProducerState.Closed)]
+        public void SetState_GivenStateIsChangedToNotWanted_ShouldNotCompleteTask(ProducerState initialState, ProducerState newState, ProducerState wantedState)
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(initialState, ProducerState.Closed);
+
+            //Act
+            var task = sut.StateChangedTo(wantedState, default);
+            _ = sut.SetState(newState);
+
+            //Assert
+            Assert.False(task.IsCompleted);
+        }
+
+        [Fact]
+        public async Task CancelToken_GivenTaskWasStillWaiting_ShouldCancelTask()
+        {
+            //Arrange
+            var sut = new StateManager<ProducerState>(ProducerState.Connected, ProducerState.Closed);
+            var cts = new CancellationTokenSource();
+            var task = sut.StateChangedFrom(ProducerState.Connected, cts.Token);
+
+            //Act
+            cts.Cancel();
+            var exception = await Record.ExceptionAsync(() => task);
+
+            //Assert
+            Assert.IsType<TaskCanceledException>(exception);
+
+            //Annihilate
+            cts.Dispose();
+        }
+    }
+}
diff --git a/src/DotPulsar.sln b/src/DotPulsar.sln
new file mode 100644
index 0000000..db0a03e
--- /dev/null
+++ b/src/DotPulsar.sln
@@ -0,0 +1,31 @@
+﻿
+Microsoft Visual Studio Solution File, Format Version 12.00
+# Visual Studio Version 16
+VisualStudioVersion = 16.0.29209.62
+MinimumVisualStudioVersion = 10.0.40219.1
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotPulsar", "DotPulsar\DotPulsar.csproj", "{8D300FC3-2E35-4D23-B0DD-FEE9E153330A}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DotPulsar.Tests", "DotPulsar.Tests\DotPulsar.Tests.csproj", "{B3FCD2D5-8009-4281-ACDB-6A7BC99606B4}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{8D300FC3-2E35-4D23-B0DD-FEE9E153330A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{8D300FC3-2E35-4D23-B0DD-FEE9E153330A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{8D300FC3-2E35-4D23-B0DD-FEE9E153330A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{8D300FC3-2E35-4D23-B0DD-FEE9E153330A}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B3FCD2D5-8009-4281-ACDB-6A7BC99606B4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B3FCD2D5-8009-4281-ACDB-6A7BC99606B4}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B3FCD2D5-8009-4281-ACDB-6A7BC99606B4}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B3FCD2D5-8009-4281-ACDB-6A7BC99606B4}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(SolutionProperties) = preSolution
+		HideSolutionNode = FALSE
+	EndGlobalSection
+	GlobalSection(ExtensibilityGlobals) = postSolution
+		SolutionGuid = {88355922-E70A-4B73-B7F8-ABF8F2B59789}
+	EndGlobalSection
+EndGlobal
diff --git a/src/DotPulsar/Abstractions/IConsumer.cs b/src/DotPulsar/Abstractions/IConsumer.cs
new file mode 100644
index 0000000..9cb5830
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IConsumer.cs
@@ -0,0 +1,52 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A consumer abstraction
+    /// </summary>
+    public interface IConsumer : IStateChanged<ConsumerState>, IDisposable
+    {
+        /// <summary>
+        /// Acknowledge the consumption of a single message
+        /// </summary>
+        Task Acknowledge(Message message, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Acknowledge the consumption of a single message using the MessageId
+        /// </summary>
+        Task Acknowledge(MessageId messageId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Acknowledge the consumption of all the messages in the topic up to and including the provided message
+        /// </summary>
+        Task AcknowledgeCumulative(Message message, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Acknowledge the consumption of all the messages in the topic up to and including the provided MessageId
+        /// </summary>
+        Task AcknowledgeCumulative(MessageId messageId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Get the MessageId of the last message on the topic
+        /// </summary>
+        Task<MessageId> GetLastMessageId(CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Receives a single message
+        /// </summary>
+        Task<Message> Receive(CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Reset the subscription associated with this consumer to a specific MessageId
+        /// </summary>
+        Task Seek(MessageId messageId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Unsubscribe the consumer
+        /// </summary>
+        Task Unsubscribe(CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IConsumerBuilder.cs b/src/DotPulsar/Abstractions/IConsumerBuilder.cs
new file mode 100644
index 0000000..338b5e8
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IConsumerBuilder.cs
@@ -0,0 +1,48 @@
+﻿namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A consumer building abstraction
+    /// </summary>
+    public interface IConsumerBuilder
+    {
+        /// <summary>
+        /// Set the consumer name. This is optional.
+        /// </summary>
+        IConsumerBuilder ConsumerName(string name);
+
+        /// <summary>
+        /// Set initial position for the subscription. Default is 'Latest'.
+        /// </summary>
+        IConsumerBuilder InitialPosition(SubscriptionInitialPosition initialPosition);
+
+        /// <summary>
+        /// Set the priority level for the shared subscription consumer. Default is 0.
+        /// </summary>
+        IConsumerBuilder PriorityLevel(int priorityLevel);
+
+        /// <summary>
+        /// Number of messages that will be prefetched. Default is 1000.
+        /// </summary>
+        IConsumerBuilder MessagePrefetchCount(uint count);
+
+        /// <summary>
+        /// Set the subscription name for this consumer. This is required.
+        /// </summary>
+        IConsumerBuilder SubscriptionName(string name);
+
+        /// <summary>
+        /// Set the subscription type for this consumer. Default is 'Exclusive'.
+        /// </summary>
+        IConsumerBuilder SubscriptionType(SubscriptionType type);
+
+        /// <summary>
+        /// Set the topic for this consumer. This is required.
+        /// </summary>
+        IConsumerBuilder Topic(string topic);
+
+        /// <summary>
+        /// Create the consumer
+        /// </summary>
+        IConsumer Create();
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IMessageBuilder.cs b/src/DotPulsar/Abstractions/IMessageBuilder.cs
new file mode 100644
index 0000000..14792ba
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IMessageBuilder.cs
@@ -0,0 +1,32 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A message building abstraction
+    /// </summary>
+    public interface IMessageBuilder
+    {
+        /// <summary>
+        /// Set the event time of the message
+        /// </summary>
+        IMessageBuilder EventTime(ulong eventTime);
+
+        /// <summary>
+        /// Add/Set a property key/value on the message
+        /// </summary>
+        IMessageBuilder Property(string key, string value);
+
+        /// <summary>
+        /// Set the sequence id of the message
+        /// </summary>
+        IMessageBuilder SequenceId(ulong sequenceId);
+
+        /// <summary>
+        /// Set the consumer name
+        /// </summary>
+        Task<MessageId> Send(ReadOnlyMemory<byte> payload, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IProducer.cs b/src/DotPulsar/Abstractions/IProducer.cs
new file mode 100644
index 0000000..b59b75b
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IProducer.cs
@@ -0,0 +1,22 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A producer abstraction
+    /// </summary>
+    public interface IProducer : IStateChanged<ProducerState>, IDisposable
+    {
+        /// <summary>
+        /// Sends a message
+        /// </summary>
+        Task<MessageId> Send(ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Sends a message and metadata
+        /// </summary>
+        Task<MessageId> Send(MessageMetadata metadata, ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IProducerBuilder.cs b/src/DotPulsar/Abstractions/IProducerBuilder.cs
new file mode 100644
index 0000000..00edff5
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IProducerBuilder.cs
@@ -0,0 +1,28 @@
+﻿namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A producer building abstraction
+    /// </summary>
+    public interface IProducerBuilder
+    {
+        /// <summary>
+        /// Set the producer name. This is optional.
+        /// </summary>
+        IProducerBuilder ProducerName(string name);
+
+        /// <summary>
+        /// Set the initial sequence id. Default is 0.
+        /// </summary>
+        IProducerBuilder InitialSequenceId(ulong initialSequenceId);
+
+        /// <summary>
+        /// Set the topic for this producer. This is required.
+        /// </summary>
+        IProducerBuilder Topic(string topic);
+
+        /// <summary>
+        /// Create the producer
+        /// </summary>
+        IProducer Create();
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IPulsarClient.cs b/src/DotPulsar/Abstractions/IPulsarClient.cs
new file mode 100644
index 0000000..9e9d9c3
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IPulsarClient.cs
@@ -0,0 +1,25 @@
+﻿using System;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A pulsar client abstraction
+    /// </summary>
+    public interface IPulsarClient : IDisposable
+    {
+        /// <summary>
+        /// Create a producer
+        /// </summary>
+        IProducer CreateProducer(ProducerOptions options);
+
+        /// <summary>
+        /// Create a consumer
+        /// </summary>
+        IConsumer CreateConsumer(ConsumerOptions options);
+
+        /// <summary>
+        /// Create a reader
+        /// </summary>
+        IReader CreateReader(ReaderOptions options);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IPulsarClientBuilder.cs b/src/DotPulsar/Abstractions/IPulsarClientBuilder.cs
new file mode 100644
index 0000000..8edafd5
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IPulsarClientBuilder.cs
@@ -0,0 +1,20 @@
+﻿using System;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A pulsar client building abstraction
+    /// </summary>
+    public interface IPulsarClientBuilder
+    {
+        /// <summary>
+        /// The service URL for the Pulsar cluster
+        /// </summary>
+        IPulsarClientBuilder ServiceUrl(Uri uri);
+
+        /// <summary>
+        /// Create the client
+        /// </summary>
+        IPulsarClient Build();
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IReader.cs b/src/DotPulsar/Abstractions/IReader.cs
new file mode 100644
index 0000000..70813fa
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IReader.cs
@@ -0,0 +1,17 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A reader abstraction
+    /// </summary>
+    public interface IReader : IStateChanged<ReaderState>, IDisposable
+    {
+        /// <summary>
+        /// Receives a single message
+        /// </summary>
+        Task<Message> Receive(CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IReaderBuilder.cs b/src/DotPulsar/Abstractions/IReaderBuilder.cs
new file mode 100644
index 0000000..3e57577
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IReaderBuilder.cs
@@ -0,0 +1,33 @@
+﻿namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A reader building abstraction
+    /// </summary>
+    public interface IReaderBuilder
+    {
+        /// <summary>
+        /// Set the reader name. This is optional.
+        /// </summary>
+        IReaderBuilder ReaderName(string name);
+
+        /// <summary>
+        /// Number of messages that will be prefetched. Default is 1000.
+        /// </summary>
+        IReaderBuilder MessagePrefetchCount(uint count);
+
+        /// <summary>
+        /// The initial reader position is set to the specified message id. This is required.
+        /// </summary>
+        IReaderBuilder StartMessageId(MessageId messageId);
+
+        /// <summary>
+        /// Set the topic for this reader. This is required.
+        /// </summary>
+        IReaderBuilder Topic(string topic);
+
+        /// <summary>
+        /// Create the reader
+        /// </summary>
+        IReader Create();
+    }
+}
diff --git a/src/DotPulsar/Abstractions/IStateChanged.cs b/src/DotPulsar/Abstractions/IStateChanged.cs
new file mode 100644
index 0000000..fec32f3
--- /dev/null
+++ b/src/DotPulsar/Abstractions/IStateChanged.cs
@@ -0,0 +1,49 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Abstractions
+{
+    /// <summary>
+    /// A state change monitoring abstraction
+    /// </summary>
+    public interface IStateChanged<TState>
+    {
+        /// <summary>
+        /// Wait for the state to change to a specific state
+        /// </summary>
+        /// <returns>
+        /// The current state
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete
+        /// </remarks>
+        Task<TState> StateChangedTo(TState state, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Wait for the state to change from a specific state
+        /// </summary>
+        /// <returns>
+        /// The current state
+        /// </returns>
+        /// <remarks>
+        /// If the state change to a final state, then all awaiting tasks will complete
+        /// </remarks>
+        Task<TState> StateChangedFrom(TState state, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Ask whether the current state is final, meaning that it will never change
+        /// </summary>
+        /// <returns>
+        /// True if it's final and False if it's not
+        /// </returns>
+        bool IsFinalState();
+
+        /// <summary>
+        /// Ask whether the provided state is final, meaning that it will never change
+        /// </summary>
+        /// <returns>
+        /// True if it's final and False if it's not
+        /// </returns>
+        bool IsFinalState(TState state);
+    }
+}
diff --git a/src/DotPulsar/ConsumerOptions.cs b/src/DotPulsar/ConsumerOptions.cs
new file mode 100644
index 0000000..65b7105
--- /dev/null
+++ b/src/DotPulsar/ConsumerOptions.cs
@@ -0,0 +1,20 @@
+﻿namespace DotPulsar
+{
+    public sealed class ConsumerOptions
+    {
+        public ConsumerOptions()
+        {
+            InitialPosition = SubscriptionInitialPosition.Latest;
+            MessagePrefetchCount = 1000;
+            SubscriptionType = SubscriptionType.Exclusive;
+        }
+
+        public string ConsumerName { get; set; }
+        public SubscriptionInitialPosition InitialPosition { get; set; }
+        public int PriorityLevel { get; set; }
+        public uint MessagePrefetchCount { get; set; }
+        public string SubscriptionName { get; set; }
+        public SubscriptionType SubscriptionType { get; set; }
+        public string Topic { get; set; }
+    }
+}
diff --git a/src/DotPulsar/ConsumerState.cs b/src/DotPulsar/ConsumerState.cs
new file mode 100644
index 0000000..35dc4fd
--- /dev/null
+++ b/src/DotPulsar/ConsumerState.cs
@@ -0,0 +1,12 @@
+﻿namespace DotPulsar
+{
+    public enum ConsumerState : byte
+    {
+        Active,
+        Closed,
+        Disconnected,
+        Faulted,
+        Inactive,
+        ReachedEndOfTopic
+    }
+}
diff --git a/src/DotPulsar/DotPulsar.csproj b/src/DotPulsar/DotPulsar.csproj
new file mode 100644
index 0000000..1fbacaf
--- /dev/null
+++ b/src/DotPulsar/DotPulsar.csproj
@@ -0,0 +1,26 @@
+﻿<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp2.2</TargetFramework>
+    <Version>0.1.0</Version>
+    <AssemblyVersion>$(Version)</AssemblyVersion>
+    <FileVersion>$(Version)</FileVersion>
+    <PackageReleaseNotes>Alpha release</PackageReleaseNotes>
+    <Description>Native .NET/C# client library for Apache Pulsar</Description>
+    <Authors>dblank</Authors>
+    <Company>Danske Commodities A/S</Company>
+    <Copyright>$(Company)</Copyright>
+    <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
+    <NoWarn>1701;1702;1591</NoWarn>
+  </PropertyGroup>
+
+  <PropertyGroup Condition="'$(Configuration)|$(Platform)'=='Release|AnyCPU'">
+    <DocumentationFile>\DotPulsar.xml</DocumentationFile>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="protobuf-net" Version="2.4.0" />
+    <PackageReference Include="System.IO.Pipelines" Version="4.5.3" />
+  </ItemGroup>
+
+</Project>
diff --git a/src/DotPulsar/Exceptions/AuthenticationException.cs b/src/DotPulsar/Exceptions/AuthenticationException.cs
new file mode 100644
index 0000000..6e9e098
--- /dev/null
+++ b/src/DotPulsar/Exceptions/AuthenticationException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class AuthenticationException : DotPulsarException
+    {
+        public AuthenticationException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/AuthorizationException.cs b/src/DotPulsar/Exceptions/AuthorizationException.cs
new file mode 100644
index 0000000..72f0dbb
--- /dev/null
+++ b/src/DotPulsar/Exceptions/AuthorizationException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class AuthorizationException : DotPulsarException
+    {
+        public AuthorizationException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ChecksumException.cs b/src/DotPulsar/Exceptions/ChecksumException.cs
new file mode 100644
index 0000000..4068590
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ChecksumException.cs
@@ -0,0 +1,9 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ChecksumException : DotPulsarException
+    {
+        public ChecksumException(string message) : base(message) { }
+
+        public ChecksumException(uint expectedChecksum, uint actualChecksum) : base($"Checksum mismatch. Excepted {expectedChecksum} but was actually {actualChecksum}") { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ConsumerAssignException.cs b/src/DotPulsar/Exceptions/ConsumerAssignException.cs
new file mode 100644
index 0000000..eb4c1f7
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ConsumerAssignException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ConsumerAssignException : DotPulsarException
+    {
+        public ConsumerAssignException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ConsumerBusyException.cs b/src/DotPulsar/Exceptions/ConsumerBusyException.cs
new file mode 100644
index 0000000..aaa1069
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ConsumerBusyException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ConsumerBusyException : DotPulsarException
+    {
+        public ConsumerBusyException(string message) : base(message) { }
+    }
+}
\ No newline at end of file
diff --git a/src/DotPulsar/Exceptions/ConsumerClosedException.cs b/src/DotPulsar/Exceptions/ConsumerClosedException.cs
new file mode 100644
index 0000000..5c714f0
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ConsumerClosedException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ConsumerClosedException : DotPulsarException
+    {
+        public ConsumerClosedException() : base("Consumer has closed") { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/DotPulsarException.cs b/src/DotPulsar/Exceptions/DotPulsarException.cs
new file mode 100644
index 0000000..a57d1a0
--- /dev/null
+++ b/src/DotPulsar/Exceptions/DotPulsarException.cs
@@ -0,0 +1,11 @@
+﻿using System;
+
+namespace DotPulsar.Exceptions
+{
+    public abstract class DotPulsarException : Exception
+    {
+        public DotPulsarException(string message) : base(message) { }
+
+        public DotPulsarException(string message, Exception innerException) : base(message, innerException) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/IncompatibleSchemaException.cs b/src/DotPulsar/Exceptions/IncompatibleSchemaException.cs
new file mode 100644
index 0000000..6deb8ce
--- /dev/null
+++ b/src/DotPulsar/Exceptions/IncompatibleSchemaException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class IncompatibleSchemaException : DotPulsarException
+    {
+        public IncompatibleSchemaException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/InvalidTopicNameException.cs b/src/DotPulsar/Exceptions/InvalidTopicNameException.cs
new file mode 100644
index 0000000..784fb2e
--- /dev/null
+++ b/src/DotPulsar/Exceptions/InvalidTopicNameException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class InvalidTopicNameException : DotPulsarException
+    {
+        public InvalidTopicNameException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/MetadataException.cs b/src/DotPulsar/Exceptions/MetadataException.cs
new file mode 100644
index 0000000..96a65fd
--- /dev/null
+++ b/src/DotPulsar/Exceptions/MetadataException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class MetadataException : DotPulsarException
+    {
+        public MetadataException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/PersistenceException.cs b/src/DotPulsar/Exceptions/PersistenceException.cs
new file mode 100644
index 0000000..918667b
--- /dev/null
+++ b/src/DotPulsar/Exceptions/PersistenceException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class PersistenceException : DotPulsarException
+    {
+        public PersistenceException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ProducerBlockedQuotaExceededException.cs b/src/DotPulsar/Exceptions/ProducerBlockedQuotaExceededException.cs
new file mode 100644
index 0000000..c72bc0b
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ProducerBlockedQuotaExceededException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ProducerBlockedQuotaExceededException : DotPulsarException
+    {
+        public ProducerBlockedQuotaExceededException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ProducerBusyException.cs b/src/DotPulsar/Exceptions/ProducerBusyException.cs
new file mode 100644
index 0000000..a465ba3
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ProducerBusyException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ProducerBusyException : DotPulsarException
+    {
+        public ProducerBusyException(string message) : base(message) { }
+    }
+}
\ No newline at end of file
diff --git a/src/DotPulsar/Exceptions/ProducerClosedException.cs b/src/DotPulsar/Exceptions/ProducerClosedException.cs
new file mode 100644
index 0000000..6fa5e4f
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ProducerClosedException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ProducerClosedException : DotPulsarException
+    {
+        public ProducerClosedException() : base("Producer has closed") { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/PulsarClientClosedException.cs b/src/DotPulsar/Exceptions/PulsarClientClosedException.cs
new file mode 100644
index 0000000..988a4b9
--- /dev/null
+++ b/src/DotPulsar/Exceptions/PulsarClientClosedException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class PulsarClientClosedException : DotPulsarException
+    {
+        public PulsarClientClosedException() : base("Client has closed") { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/ReaderClosedException.cs b/src/DotPulsar/Exceptions/ReaderClosedException.cs
new file mode 100644
index 0000000..0d08ed8
--- /dev/null
+++ b/src/DotPulsar/Exceptions/ReaderClosedException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class ReaderClosedException : DotPulsarException
+    {
+        public ReaderClosedException() : base("Reader has closed") { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/SubscriptionNotFoundException.cs b/src/DotPulsar/Exceptions/SubscriptionNotFoundException.cs
new file mode 100644
index 0000000..da3eaba
--- /dev/null
+++ b/src/DotPulsar/Exceptions/SubscriptionNotFoundException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class SubscriptionNotFoundException : DotPulsarException
+    {
+        public SubscriptionNotFoundException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/TopicNotFoundException.cs b/src/DotPulsar/Exceptions/TopicNotFoundException.cs
new file mode 100644
index 0000000..d10a36c
--- /dev/null
+++ b/src/DotPulsar/Exceptions/TopicNotFoundException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class TopicNotFoundException : DotPulsarException
+    {
+        public TopicNotFoundException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/TopicTerminatedException.cs b/src/DotPulsar/Exceptions/TopicTerminatedException.cs
new file mode 100644
index 0000000..2222040
--- /dev/null
+++ b/src/DotPulsar/Exceptions/TopicTerminatedException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class TopicTerminatedException : DotPulsarException
+    {
+        public TopicTerminatedException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/UnknownException.cs b/src/DotPulsar/Exceptions/UnknownException.cs
new file mode 100644
index 0000000..bd2772c
--- /dev/null
+++ b/src/DotPulsar/Exceptions/UnknownException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class UnknownException : DotPulsarException
+    {
+        public UnknownException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Exceptions/UnsupportedVersionException.cs b/src/DotPulsar/Exceptions/UnsupportedVersionException.cs
new file mode 100644
index 0000000..cdcb833
--- /dev/null
+++ b/src/DotPulsar/Exceptions/UnsupportedVersionException.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Exceptions
+{
+    public sealed class UnsupportedVersionException : DotPulsarException
+    {
+        public UnsupportedVersionException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Extensions/ProducerExtensions.cs b/src/DotPulsar/Extensions/ProducerExtensions.cs
new file mode 100644
index 0000000..c482522
--- /dev/null
+++ b/src/DotPulsar/Extensions/ProducerExtensions.cs
@@ -0,0 +1,10 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Internal;
+
+namespace DotPulsar.Extensions
+{
+    public static class ProducerExtensions
+    {
+        public static IMessageBuilder NewMessage(this IProducer producer) => new MessageBuilder(producer);
+    }
+}
diff --git a/src/DotPulsar/Extensions/PulsarClientExtensions.cs b/src/DotPulsar/Extensions/PulsarClientExtensions.cs
new file mode 100644
index 0000000..025f207
--- /dev/null
+++ b/src/DotPulsar/Extensions/PulsarClientExtensions.cs
@@ -0,0 +1,12 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Internal;
+
+namespace DotPulsar.Extensions
+{
+    public static class PulsarClientExtensions
+    {
+        public static IProducerBuilder NewProducer(this IPulsarClient pulsarClient) => new ProducerBuilder(pulsarClient);
+        public static IConsumerBuilder NewConsumer(this IPulsarClient pulsarClient) => new ConsumerBuilder(pulsarClient);
+        public static IReaderBuilder NewReader(this IPulsarClient pulsarClient) => new ReaderBuilder(pulsarClient);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IConsumerProxy.cs b/src/DotPulsar/Internal/Abstractions/IConsumerProxy.cs
new file mode 100644
index 0000000..4025088
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IConsumerProxy.cs
@@ -0,0 +1,10 @@
+﻿namespace DotPulsar.Internal.Abstractions
+{
+    public interface IConsumerProxy : IEnqueue<MessagePackage>, IDequeue<MessagePackage>
+    {
+        void Active();
+        void Inactive();
+        void Disconnected();
+        void ReachedEndOfTopic();
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IConsumerStream.cs b/src/DotPulsar/Internal/Abstractions/IConsumerStream.cs
new file mode 100644
index 0000000..f3443ed
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IConsumerStream.cs
@@ -0,0 +1,16 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IConsumerStream : IDisposable
+    {
+        Task Send(CommandAck command);
+        Task<CommandSuccess> Send(CommandUnsubscribe command);
+        Task<CommandSuccess> Send(CommandSeek command);
+        Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command);
+        Task<Message> Receive(CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IConsumerStreamFactory.cs b/src/DotPulsar/Internal/Abstractions/IConsumerStreamFactory.cs
new file mode 100644
index 0000000..b8767ff
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IConsumerStreamFactory.cs
@@ -0,0 +1,10 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IConsumerStreamFactory
+    {
+        Task<IConsumerStream> CreateStream(IConsumerProxy proxy, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IDequeue.cs b/src/DotPulsar/Internal/Abstractions/IDequeue.cs
new file mode 100644
index 0000000..a456fed
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IDequeue.cs
@@ -0,0 +1,10 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IDequeue<T>
+    {
+        Task<T> Dequeue(CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IEnqueue.cs b/src/DotPulsar/Internal/Abstractions/IEnqueue.cs
new file mode 100644
index 0000000..100485d
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IEnqueue.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Internal.Abstractions
+{
+    public interface IEnqueue<T>
+    {
+        void Enqueue(T item);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IFaultStrategy.cs b/src/DotPulsar/Internal/Abstractions/IFaultStrategy.cs
new file mode 100644
index 0000000..b3e141e
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IFaultStrategy.cs
@@ -0,0 +1,10 @@
+﻿using System;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IFaultStrategy
+    {
+        FaultAction DetermineFaultAction(Exception exception);
+        TimeSpan TimeToWait { get; }
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IProducerProxy.cs b/src/DotPulsar/Internal/Abstractions/IProducerProxy.cs
new file mode 100644
index 0000000..e6823f4
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IProducerProxy.cs
@@ -0,0 +1,7 @@
+﻿namespace DotPulsar.Internal.Abstractions
+{
+    public interface IProducerProxy
+    {
+        void Disconnected();
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IProducerStream.cs b/src/DotPulsar/Internal/Abstractions/IProducerStream.cs
new file mode 100644
index 0000000..a9e94c2
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IProducerStream.cs
@@ -0,0 +1,11 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IProducerStream : IDisposable
+    {
+        Task<CommandSendReceipt> Send(PulsarApi.MessageMetadata metadata, ReadOnlyMemory<byte> payload);
+    }
+}
diff --git a/src/DotPulsar/Internal/Abstractions/IProducerStreamFactory.cs b/src/DotPulsar/Internal/Abstractions/IProducerStreamFactory.cs
new file mode 100644
index 0000000..d0a5f06
--- /dev/null
+++ b/src/DotPulsar/Internal/Abstractions/IProducerStreamFactory.cs
@@ -0,0 +1,10 @@
+﻿using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal.Abstractions
+{
+    public interface IProducerStreamFactory
+    {
+        Task<IProducerStream> CreateStream(IProducerProxy proxy, CancellationToken cancellationToken = default);
+    }
+}
diff --git a/src/DotPulsar/Internal/AsyncLock.cs b/src/DotPulsar/Internal/AsyncLock.cs
new file mode 100644
index 0000000..b7d0b4f
--- /dev/null
+++ b/src/DotPulsar/Internal/AsyncLock.cs
@@ -0,0 +1,116 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class AsyncLock : IDisposable
+    {
+        private readonly LinkedList<CancelableCompletionSource<IDisposable>> _pending;
+        private readonly SemaphoreSlim _semaphoreSlim;
+        private readonly Releaser _releaser;
+        private readonly Task<IDisposable> _completedTask;
+        private bool _isDisposed;
+
+        public AsyncLock()
+        {
+            _pending = new LinkedList<CancelableCompletionSource<IDisposable>>();
+            _semaphoreSlim = new SemaphoreSlim(1, 1);
+            _releaser = new Releaser(Release);
+            _completedTask = Task.FromResult((IDisposable)_releaser);
+            _isDisposed = false;
+        }
+
+        public Task<IDisposable> Lock(CancellationToken cancellationToken = default)
+        {
+            LinkedListNode<CancelableCompletionSource<IDisposable>> node = null;
+
+            lock (_pending)
+            {
+                if (_isDisposed)
+                    throw new ObjectDisposedException(nameof(AsyncLock));
+
+                if (_semaphoreSlim.CurrentCount == 1) //Lock is free
+                {
+                    _semaphoreSlim.Wait(); //Will never block
+                    return _completedTask;
+                }
+
+                //Lock was not free
+                var ccs = new CancelableCompletionSource<IDisposable>();
+                node = _pending.AddLast(ccs);
+            }
+
+            cancellationToken.Register(() => Cancel(node));
+
+            return node.Value.Task;
+        }
+
+        public void Dispose()
+        {
+            lock (_pending)
+            {
+                if (_isDisposed)
+                    return;
+
+                _isDisposed = true;
+
+                foreach (var pending in _pending)
+                {
+                    pending.SetException(new ObjectDisposedException(nameof(AsyncLock)));
+                    pending.Dispose();
+                }
+
+                _pending.Clear();
+            }
+
+            _semaphoreSlim.Wait(); //Wait for possible lock-holder to finish
+            _semaphoreSlim.Release();
+            _semaphoreSlim.Dispose();
+        }
+
+        private void Cancel(LinkedListNode<CancelableCompletionSource<IDisposable>> node)
+        {
+            lock (_pending)
+            {
+                try
+                {
+                    _pending.Remove(node);
+                    node.Value.Dispose();
+                }
+                catch
+                {
+                    // Ignore
+                }
+            }
+        }
+
+        private void Release()
+        {
+            lock (_pending)
+            {
+                if (_pending.Count > 0)
+                {
+                    var node = _pending.First;
+                    node.Value.SetResult(_releaser);
+                    node.Value.Dispose();
+                    _pending.RemoveFirst();
+                    return;
+                }
+
+                if (_semaphoreSlim.CurrentCount == 0)
+                    _semaphoreSlim.Release();
+            }
+        }
+
+        private class Releaser : IDisposable
+        {
+            private readonly Action _release;
+
+            public Releaser(Action release) => _release = release;
+
+            public void Dispose() => _release();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/AsyncQueue.cs b/src/DotPulsar/Internal/AsyncQueue.cs
new file mode 100644
index 0000000..4dbf084
--- /dev/null
+++ b/src/DotPulsar/Internal/AsyncQueue.cs
@@ -0,0 +1,78 @@
+﻿using DotPulsar.Internal.Abstractions;
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class AsyncQueue<T> : IEnqueue<T>, IDequeue<T>, IDisposable
+    {
+        private readonly object _lock;
+        private readonly Queue<T> _queue;
+        private readonly LinkedList<CancelableCompletionSource<T>> _pendingDequeues;
+
+        public AsyncQueue()
+        {
+            _lock = new object();
+            _queue = new Queue<T>();
+            _pendingDequeues = new LinkedList<CancelableCompletionSource<T>>();
+        }
+
+        public void Enqueue(T item)
+        {
+            lock (_lock)
+            {
+                if (_pendingDequeues.Count > 0)
+                {
+                    var tcs = _pendingDequeues.First;
+                    _pendingDequeues.RemoveFirst();
+                    tcs.Value.SetResult(item);
+                }
+                else
+                    _queue.Enqueue(item);
+            }
+        }
+
+        public Task<T> Dequeue(CancellationToken cancellationToken = default)
+        {
+            LinkedListNode<CancelableCompletionSource<T>> node = null;
+
+            lock (_lock)
+            {
+                if (_queue.Count > 0)
+                    return Task.FromResult(_queue.Dequeue());
+
+                node = _pendingDequeues.AddLast(new CancelableCompletionSource<T>());
+            }
+
+            node.Value.SetupCancellation(() => Cancel(node), cancellationToken);
+            return node.Value.Task;
+        }
+
+        public void Dispose()
+        {
+            lock (_lock)
+            {
+                foreach (var pendingDequeue in _pendingDequeues)
+                    pendingDequeue.Dispose();
+
+                _pendingDequeues.Clear();
+                _queue.Clear();
+            }
+        }
+
+        private void Cancel(LinkedListNode<CancelableCompletionSource<T>> node)
+        {
+            lock (_lock)
+            {
+                try
+                {
+                    node.Value.Dispose();
+                    _pendingDequeues.Remove(node);
+                }
+                catch { }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Awaitor.cs b/src/DotPulsar/Internal/Awaitor.cs
new file mode 100644
index 0000000..83e1143
--- /dev/null
+++ b/src/DotPulsar/Internal/Awaitor.cs
@@ -0,0 +1,36 @@
+﻿using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Awaitor<T, Result> : IDisposable
+    {
+        private readonly Dictionary<T, TaskCompletionSource<Result>> _items;
+
+        public Awaitor() => _items = new Dictionary<T, TaskCompletionSource<Result>>();
+
+        public Task<Result> CreateTask(T item)
+        {
+            var tcs = new TaskCompletionSource<Result>(TaskCreationOptions.RunContinuationsAsynchronously);
+            _items.Add(item, tcs);
+            return tcs.Task;
+        }
+
+        public void SetResult(T item, Result result)
+        {
+            _items.Remove(item, out var tcs);
+            tcs.SetResult(result);
+        }
+
+        public void Dispose()
+        {
+            foreach (var item in _items.Values)
+            {
+                item.SetCanceled();
+            }
+
+            _items.Clear();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/CancelableCompletionSource.cs b/src/DotPulsar/Internal/CancelableCompletionSource.cs
new file mode 100644
index 0000000..ebf87d4
--- /dev/null
+++ b/src/DotPulsar/Internal/CancelableCompletionSource.cs
@@ -0,0 +1,28 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class CancelableCompletionSource<T> : IDisposable
+    {
+        private readonly TaskCompletionSource<T> _source;
+        private CancellationTokenRegistration? _registration;
+
+        public CancelableCompletionSource() => _source = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public void SetupCancellation(Action callback, CancellationToken token) => _registration = token.Register(() => callback());
+
+        public void SetResult(T result) => _ = _source.TrySetResult(result);
+
+        public void SetException(Exception exception) => _ = _source.TrySetException(exception);
+
+        public Task<T> Task => _source.Task;
+
+        public void Dispose()
+        {
+            _ = _source.TrySetCanceled();
+            _registration?.Dispose();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Connection.cs b/src/DotPulsar/Internal/Connection.cs
new file mode 100644
index 0000000..b5eec21
--- /dev/null
+++ b/src/DotPulsar/Internal/Connection.cs
@@ -0,0 +1,191 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.Extensions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Buffers;
+using System.IO;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Connection : IDisposable
+    {
+        private readonly AsyncLock _lock;
+        private readonly ProducerManager _producerManager;
+        private readonly ConsumerManager _consumerManager;
+        private readonly RequestResponseHandler _requestResponseHandler;
+        private readonly PingPongHandler _pingPongHandler;
+        private readonly PulsarStream _stream;
+
+        public Connection(Stream stream)
+        {
+            _lock = new AsyncLock();
+            _producerManager = new ProducerManager();
+            _consumerManager = new ConsumerManager();
+            _requestResponseHandler = new RequestResponseHandler();
+            _pingPongHandler = new PingPongHandler(this);
+            _stream = new PulsarStream(stream, HandleCommand);
+        }
+
+        public Task IsClosed => _stream.IsClosed;
+
+        public async Task<bool> IsActive()
+        {
+            using (await _lock.Lock())
+            {
+                return _producerManager.HasProducers || _consumerManager.HasConsumers;
+            }
+        }
+
+        public async Task<ProducerResponse> Send(CommandProducer command, IProducerProxy proxy)
+        {
+            Task<BaseCommand> responseTask = null;
+            using (await _lock.Lock())
+            {
+                _producerManager.Outgoing(command, proxy);
+                var baseCommand = command.AsBaseCommand();
+                responseTask = _requestResponseHandler.Outgoing(baseCommand);
+                var sequence = Serializer.Serialize(baseCommand);
+                await _stream.Send(sequence);
+            }
+
+            var response = await responseTask;
+            if (response.CommandType == BaseCommand.Type.Error)
+            {
+                _producerManager.Remove(command.ProducerId);
+                response.Error.Throw();
+            }
+
+            return new ProducerResponse(command.ProducerId, response.ProducerSuccess.ProducerName);
+        }
+
+        public async Task<SubscribeResponse> Send(CommandSubscribe command, IConsumerProxy proxy)
+        {
+            Task<BaseCommand> responseTask = null;
+            using (await _lock.Lock())
+            {
+                _consumerManager.Outgoing(command, proxy);
+                var baseCommand = command.AsBaseCommand();
+                responseTask = _requestResponseHandler.Outgoing(baseCommand);
+                var sequence = Serializer.Serialize(baseCommand);
+                await _stream.Send(sequence);
+            }
+
+            var response = await responseTask;
+            if (response.CommandType == BaseCommand.Type.Error)
+            {
+                _consumerManager.Remove(command.ConsumerId);
+                response.Error.Throw();
+            }
+
+            return new SubscribeResponse(command.ConsumerId);
+        }
+
+        public async Task Send(CommandPing command) => await Send(command.AsBaseCommand());
+        public async Task Send(CommandPong command) => await Send(command.AsBaseCommand());
+        public async Task Send(CommandAck command) => await Send(command.AsBaseCommand());
+        public async Task Send(CommandFlow command) => await Send(command.AsBaseCommand());
+
+        public async Task<BaseCommand> Send(CommandUnsubscribe command)
+        {
+            var response = await SendRequestResponse(command.AsBaseCommand());
+            if (response.CommandType == BaseCommand.Type.Success)
+                _consumerManager.Remove(command.ConsumerId);
+            return response;
+        }
+
+        public async Task<BaseCommand> Send(CommandConnect command) => await SendRequestResponse(command.AsBaseCommand());
+        public async Task<BaseCommand> Send(CommandLookupTopic command) => await SendRequestResponse(command.AsBaseCommand());
+        public async Task<BaseCommand> Send(CommandSeek command) => await SendRequestResponse(command.AsBaseCommand());
+        public async Task<BaseCommand> Send(CommandGetLastMessageId command) => await SendRequestResponse(command.AsBaseCommand());
+
+        public async Task<BaseCommand> Send(CommandCloseProducer command)
+        {
+            var response = await SendRequestResponse(command.AsBaseCommand());
+            if (response.CommandType == BaseCommand.Type.Success)
+                _producerManager.Remove(command.ProducerId);
+            return response;
+        }
+
+        public async Task<BaseCommand> Send(CommandCloseConsumer command)
+        {
+            var response = await SendRequestResponse(command.AsBaseCommand());
+            if (response.CommandType == BaseCommand.Type.Success)
+                _consumerManager.Remove(command.ConsumerId);
+            return response;
+        }
+
+        public async Task<BaseCommand> Send(SendPackage command)
+        {
+            Task<BaseCommand> response = null;
+            using (await _lock.Lock())
+            {
+                var baseCommand = command.Command.AsBaseCommand();
+                response = _requestResponseHandler.Outgoing(baseCommand);
+                var sequence = Serializer.Serialize(baseCommand, command.Metadata, command.Payload);
+                await _stream.Send(sequence);
+            }
+            return await response;
+        }
+
+        private async Task<BaseCommand> SendRequestResponse(BaseCommand command)
+        {
+            Task<BaseCommand> response = null;
+            using (await _lock.Lock())
+            {
+                response = _requestResponseHandler.Outgoing(command);
+                var sequence = Serializer.Serialize(command);
+                await _stream.Send(sequence);
+            }
+            return await response;
+        }
+
+        private async Task Send(BaseCommand command)
+        {
+            using (await _lock.Lock())
+            {
+                var sequence = Serializer.Serialize(command);
+                await _stream.Send(sequence);
+            }
+        }
+
+        private void HandleCommand(uint commandSize, ReadOnlySequence<byte> sequence)
+        {
+            var command = Serializer.Deserialize<BaseCommand>(sequence.Slice(0, commandSize));
+
+            switch (command.CommandType)
+            {
+                case BaseCommand.Type.Message:
+                    _consumerManager.Incoming(new MessagePackage(command.Message, sequence.Slice(commandSize)));
+                    return;
+                case BaseCommand.Type.CloseConsumer:
+                    _consumerManager.Incoming(command.CloseConsumer);
+                    return;
+                case BaseCommand.Type.ActiveConsumerChange:
+                    _consumerManager.Incoming(command.ActiveConsumerChange);
+                    return;
+                case BaseCommand.Type.ReachedEndOfTopic:
+                    _consumerManager.Incoming(command.ReachedEndOfTopic);
+                    return;
+                case BaseCommand.Type.CloseProducer:
+                    _producerManager.Incoming(command.CloseProducer);
+                    return;
+                case BaseCommand.Type.Ping:
+                    _pingPongHandler.Incoming(command.Ping);
+                    return;
+                default:
+                    _requestResponseHandler.Incoming(command);
+                    return;
+            }
+        }
+
+        public void Dispose()
+        {
+            _lock.Dispose();
+            _consumerManager.Dispose();
+            _producerManager.Dispose();
+            _requestResponseHandler.Dispose();
+            _stream.Dispose();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ConnectionPool.cs b/src/DotPulsar/Internal/ConnectionPool.cs
new file mode 100644
index 0000000..de3ac0b
--- /dev/null
+++ b/src/DotPulsar/Internal/ConnectionPool.cs
@@ -0,0 +1,143 @@
+﻿using DotPulsar.Internal.Extensions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Collections.Concurrent;
+using System.ComponentModel;
+using System.Linq;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConnectionPool : IDisposable
+    {
+        private readonly AsyncLock _lock;
+        private readonly int _protocolVersion;
+        private readonly string _clientVersion;
+        private readonly Uri _serviceUrl;
+        private readonly ConcurrentDictionary<Uri, Connection> _connections;
+        private readonly CancellationTokenSource _cancellationTokenSource;
+        private readonly Task _closeInactiveConnections;
+
+        public ConnectionPool(int protocolVersion, string clientVersion, Uri serviceUrl)
+        {
+            _lock = new AsyncLock();
+            _protocolVersion = protocolVersion;
+            _clientVersion = clientVersion;
+            _serviceUrl = serviceUrl;
+            _connections = new ConcurrentDictionary<Uri, Connection>();
+            _cancellationTokenSource = new CancellationTokenSource();
+            _closeInactiveConnections = CloseInactiveConnections(TimeSpan.FromSeconds(60), _cancellationTokenSource.Token);
+        }
+
+        public void Dispose() //While we wait for IAsyncDisposable
+        {
+            _cancellationTokenSource.Cancel();
+            _closeInactiveConnections.Wait();
+
+            _lock.Dispose();
+
+            foreach (var serviceUrl in _connections.Keys.ToArray())
+            {
+                Deregister(serviceUrl);
+            }
+        }
+
+        public async Task<Connection> FindConnectionForTopic(string topic, CancellationToken cancellationToken)
+        {
+            var connection = await CreateConnection(_serviceUrl, cancellationToken);
+
+            var authoritative = false;
+
+            while (true)
+            {
+                var lookup = new CommandLookupTopic
+                {
+                    Topic = topic,
+                    Authoritative = authoritative
+                };
+                var response = await connection.Send(lookup);
+                response.Expect(BaseCommand.Type.LookupResponse);
+
+                switch (response.LookupTopicResponse.Response)
+                {
+                    case CommandLookupTopicResponse.LookupType.Connect:
+                        return connection;
+                    case CommandLookupTopicResponse.LookupType.Redirect:
+                        authoritative = response.LookupTopicResponse.Authoritative;
+                        connection = await CreateConnection(new Uri(response.LookupTopicResponse.BrokerServiceUrl), cancellationToken);
+                        continue;
+                    case CommandLookupTopicResponse.LookupType.Failed:
+                        response.LookupTopicResponse.Throw();
+                        break;
+                    default:
+                        throw new InvalidEnumArgumentException("LookupType", (int)response.LookupTopicResponse.Response, typeof(CommandLookupTopicResponse.LookupType));
+                }
+            }
+        }
+
+        private async Task<Connection> CreateConnection(Uri serviceUrl, CancellationToken cancellationToken)
+        {
+            using (await _lock.Lock(cancellationToken))
+            {
+                if (_connections.TryGetValue(serviceUrl, out Connection connection))
+                    return connection;
+
+                var tcpClient = new TcpClient();
+                await tcpClient.ConnectAsync(serviceUrl.Host, serviceUrl.Port);
+
+                connection = new Connection(tcpClient.GetStream());
+                Register(serviceUrl, connection);
+
+                var connect = new CommandConnect
+                {
+                    ProtocolVersion = _protocolVersion,
+                    ClientVersion = _clientVersion
+                };
+
+                var response = await connection.Send(connect);
+                response.Expect(BaseCommand.Type.Connected);
+
+                return connection;
+            }
+        }
+
+        private void Register(Uri serviceUrl, Connection connection)
+        {
+            _connections[serviceUrl] = connection;
+            connection.IsClosed.ContinueWith(t => Deregister(serviceUrl));
+        }
+
+        private void Deregister(Uri serviceUrl)
+        {
+            if (_connections.TryRemove(serviceUrl, out Connection connection))
+                connection.Dispose();
+        }
+
+        private async Task CloseInactiveConnections(TimeSpan interval, CancellationToken cancellationToken)
+        {
+            while (!cancellationToken.IsCancellationRequested)
+            {
+                try
+                {
+                    await Task.Delay(interval, cancellationToken);
+
+                    using (await _lock.Lock(cancellationToken))
+                    {
+                        var serviceUrls = _connections.Keys;
+                        foreach (var serviceUrl in serviceUrls)
+                        {
+                            var connection = _connections[serviceUrl];
+                            if (connection == null)
+                                continue;
+                            if (!await connection.IsActive())
+                                Deregister(serviceUrl);
+                        }
+                    }
+                }
+                catch { }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Consumer.cs b/src/DotPulsar/Internal/Consumer.cs
new file mode 100644
index 0000000..3317313
--- /dev/null
+++ b/src/DotPulsar/Internal/Consumer.cs
@@ -0,0 +1,156 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Exceptions;
+using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Consumer : IConsumer
+    {
+        private readonly Executor _executor;
+        private readonly CommandAck _cachedCommandAck;
+        private readonly IConsumerStreamFactory _streamFactory;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly bool _setProxyState;
+        private readonly StateManager<ConsumerState> _stateManager;
+        private readonly CancellationTokenSource _connectTokenSource;
+        private readonly Task _connectTask;
+        private Action _throwIfClosedOrFaulted;
+        private IConsumerStream Stream { get; set; }
+
+        public Consumer(IConsumerStreamFactory streamFactory, IFaultStrategy faultStrategy, bool setProxyState)
+        {
+            _executor = new Executor(ExecutorOnException);
+            _cachedCommandAck = new CommandAck();
+            _stateManager = new StateManager<ConsumerState>(ConsumerState.Disconnected, ConsumerState.Closed, ConsumerState.ReachedEndOfTopic, ConsumerState.Faulted);
+            _streamFactory = streamFactory;
+            _faultStrategy = faultStrategy;
+            _setProxyState = setProxyState;
+            _connectTokenSource = new CancellationTokenSource();
+            Stream = new NotReadyStream();
+            _connectTask = Connect(_connectTokenSource.Token);
+            _throwIfClosedOrFaulted = () => { };
+        }
+
+        public async Task<ConsumerState> StateChangedTo(ConsumerState state, CancellationToken cancellationToken) => await _stateManager.StateChangedTo(state, cancellationToken);
+        public async Task<ConsumerState> StateChangedFrom(ConsumerState state, CancellationToken cancellationToken) => await _stateManager.StateChangedFrom(state, cancellationToken);
+        public bool IsFinalState() => _stateManager.IsFinalState();
+        public bool IsFinalState(ConsumerState state) => _stateManager.IsFinalState(state);
+
+        public void Dispose()
+        {
+            _executor.Dispose();
+            _connectTokenSource.Cancel();
+            _connectTask.Wait();
+        }
+
+        public async Task<Message> Receive(CancellationToken cancellationToken) => await _executor.Execute(() => Stream.Receive(cancellationToken), cancellationToken);
+
+        public async Task Acknowledge(Message message, CancellationToken cancellationToken)
+            => await Acknowledge(message.MessageId.Data, CommandAck.AckType.Individual, cancellationToken);
+
+        public async Task Acknowledge(MessageId messageId, CancellationToken cancellationToken)
+            => await Acknowledge(messageId.Data, CommandAck.AckType.Individual, cancellationToken);
+
+        public async Task AcknowledgeCumulative(Message message, CancellationToken cancellationToken)
+            => await Acknowledge(message.MessageId.Data, CommandAck.AckType.Cumulative, cancellationToken);
+
+        public async Task AcknowledgeCumulative(MessageId messageId, CancellationToken cancellationToken)
+            => await Acknowledge(messageId.Data, CommandAck.AckType.Cumulative, cancellationToken);
+
+        public async Task Unsubscribe(CancellationToken cancellationToken)
+        {
+            _ = await _executor.Execute(() => Stream.Send(new CommandUnsubscribe()), cancellationToken);
+            HasClosed();
+        }
+
+        public async Task Seek(MessageId messageId, CancellationToken cancellationToken)
+        {
+            var seek = new CommandSeek { MessageId = messageId.Data };
+            _ = await _executor.Execute(() => Stream.Send(seek), cancellationToken);
+            return;
+        }
+
+        public async Task<MessageId> GetLastMessageId(CancellationToken cancellationToken)
+        {
+            var response = await _executor.Execute(() => Stream.Send(new CommandGetLastMessageId()), cancellationToken);
+            return new MessageId(response.LastMessageId);
+        }
+
+        private async Task Acknowledge(MessageIdData messageIdData, CommandAck.AckType ackType, CancellationToken cancellationToken)
+        {
+            await _executor.Execute(() =>
+            {
+                _cachedCommandAck.Type = ackType;
+                _cachedCommandAck.MessageIds.Clear();
+                _cachedCommandAck.MessageIds.Add(messageIdData);
+                return Stream.Send(_cachedCommandAck);
+            }, cancellationToken);
+        }
+
+        private async Task ExecutorOnException(Exception exception, CancellationToken cancellationToken)
+        {
+            _throwIfClosedOrFaulted();
+
+            switch (_faultStrategy.DetermineFaultAction(exception))
+            {
+                case FaultAction.Retry:
+                    await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                    break;
+                case FaultAction.Relookup:
+                    await _stateManager.StateChangedFrom(ConsumerState.Disconnected, cancellationToken);
+                    break;
+                case FaultAction.Fault:
+                    HasFaulted(exception);
+                    break;
+            }
+
+            _throwIfClosedOrFaulted();
+        }
+
+        private void HasFaulted(Exception exception)
+        {
+            _throwIfClosedOrFaulted = () => throw exception;
+            _stateManager.SetState(ConsumerState.Faulted);
+        }
+
+        private void HasClosed()
+        {
+            _throwIfClosedOrFaulted = () => throw new ConsumerClosedException();
+            _stateManager.SetState(ConsumerState.Closed);
+        }
+
+        private async Task Connect(CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (true)
+                {
+                    using (var proxy = new ConsumerProxy(_stateManager, new AsyncQueue<MessagePackage>()))
+                    using (Stream = await _streamFactory.CreateStream(proxy, cancellationToken))
+                    {
+                        if (_setProxyState)
+                            proxy.Active();
+                        else
+                            await _stateManager.StateChangedFrom(ConsumerState.Disconnected, cancellationToken);
+
+                        await _stateManager.StateChangedTo(ConsumerState.Disconnected, cancellationToken);
+                        if (_stateManager.IsFinalState())
+                            return;
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                HasClosed();
+            }
+            catch (Exception exception)
+            {
+                HasFaulted(exception);
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ConsumerBuilder.cs b/src/DotPulsar/Internal/ConsumerBuilder.cs
new file mode 100644
index 0000000..522240f
--- /dev/null
+++ b/src/DotPulsar/Internal/ConsumerBuilder.cs
@@ -0,0 +1,60 @@
+﻿using DotPulsar.Abstractions;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConsumerBuilder : IConsumerBuilder
+    {
+        private readonly IPulsarClient _pulsarClient;
+        private readonly ConsumerOptions _options;
+
+        public ConsumerBuilder(IPulsarClient pulsarClient)
+        {
+            _pulsarClient = pulsarClient;
+            _options = new ConsumerOptions();
+        }
+
+        public IConsumerBuilder ConsumerName(string name)
+        {
+            _options.ConsumerName = name;
+            return this;
+        }
+
+        public IConsumerBuilder InitialPosition(SubscriptionInitialPosition initialPosition)
+        {
+            _options.InitialPosition = initialPosition;
+            return this;
+        }
+
+        public IConsumerBuilder PriorityLevel(int priorityLevel)
+        {
+            _options.PriorityLevel = priorityLevel;
+            return this;
+        }
+
+        public IConsumerBuilder MessagePrefetchCount(uint count)
+        {
+            _options.MessagePrefetchCount = count;
+            return this;
+        }
+
+        public IConsumerBuilder SubscriptionName(string name)
+        {
+            _options.SubscriptionName = name;
+            return this;
+        }
+
+        public IConsumerBuilder SubscriptionType(SubscriptionType type)
+        {
+            _options.SubscriptionType = type;
+            return this;
+        }
+
+        public IConsumerBuilder Topic(string topic)
+        {
+            _options.Topic = topic;
+            return this;
+        }
+
+        public IConsumer Create() => _pulsarClient.CreateConsumer(_options);
+    }
+}
diff --git a/src/DotPulsar/Internal/ConsumerManager.cs b/src/DotPulsar/Internal/ConsumerManager.cs
new file mode 100644
index 0000000..c461283
--- /dev/null
+++ b/src/DotPulsar/Internal/ConsumerManager.cs
@@ -0,0 +1,61 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConsumerManager : IDisposable
+    {
+        private readonly IdLookup<IConsumerProxy> _proxies;
+
+        public ConsumerManager() => _proxies = new IdLookup<IConsumerProxy>();
+
+        public bool HasConsumers => !_proxies.IsEmpty();
+
+        public void Outgoing(CommandSubscribe subscribe, IConsumerProxy proxy) => subscribe.ConsumerId = _proxies.Add(proxy);
+
+        public void Dispose()
+        {
+            foreach (var id in _proxies.AllIds())
+            {
+                RemoveConsumer(id);
+            }
+        }
+
+        public void Incoming(MessagePackage package)
+        {
+            var consumerId = package.Command.ConsumerId;
+            var proxy = _proxies[consumerId];
+            proxy?.Enqueue(package);
+        }
+
+        public void Incoming(CommandCloseConsumer command) => RemoveConsumer(command.ConsumerId);
+
+        public void Incoming(CommandActiveConsumerChange command)
+        {
+            var proxy = _proxies[command.ConsumerId];
+            if (proxy == null) return;
+
+            if (command.IsActive)
+                proxy.Active();
+            else
+                proxy.Inactive();
+        }
+
+        public void Incoming(CommandReachedEndOfTopic command)
+        {
+            var proxy = _proxies[command.ConsumerId];
+            proxy?.ReachedEndOfTopic();
+        }
+
+        public void Remove(ulong consumerId) => _proxies.Remove(consumerId);
+
+        private void RemoveConsumer(ulong consumerId)
+        {
+            var proxy = _proxies[consumerId];
+            if (proxy == null) return;
+            proxy.Disconnected();
+            _proxies.Remove(consumerId);
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ConsumerProxy.cs b/src/DotPulsar/Internal/ConsumerProxy.cs
new file mode 100644
index 0000000..8d87a48
--- /dev/null
+++ b/src/DotPulsar/Internal/ConsumerProxy.cs
@@ -0,0 +1,54 @@
+﻿using DotPulsar.Internal.Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConsumerProxy : IConsumerProxy, IDisposable
+    {
+        private readonly object _lock;
+        private readonly StateManager<ConsumerState> _stateManager;
+        private readonly AsyncQueue<MessagePackage> _queue;
+        private bool _hasDisconnected;
+
+        public ConsumerProxy(StateManager<ConsumerState> stateManager, AsyncQueue<MessagePackage> queue)
+        {
+            _lock = new object();
+            _stateManager = stateManager;
+            _queue = queue;
+            _hasDisconnected = false;
+        }
+
+        public void Active() => SetState(ConsumerState.Active);
+        public void Inactive() => SetState(ConsumerState.Inactive);
+        public void ReachedEndOfTopic() => SetState(ConsumerState.ReachedEndOfTopic);
+
+        public void Disconnected()
+        {
+            lock (_lock)
+            {
+                if (_hasDisconnected)
+                    return;
+
+                _stateManager.SetState(ConsumerState.Disconnected);
+                _hasDisconnected = true;
+            }
+        }
+
+
+        public void Enqueue(MessagePackage package) => _queue.Enqueue(package);
+        public async Task<MessagePackage> Dequeue(CancellationToken cancellationToken) => await _queue.Dequeue(cancellationToken);
+
+        private void SetState(ConsumerState state)
+        {
+            lock (_lock)
+            {
+                if (!_hasDisconnected)
+                    _stateManager.SetState(state);
+            }
+        }
+
+        public void Dispose() => _queue.Dispose();
+    }
+}
diff --git a/src/DotPulsar/Internal/ConsumerStream.cs b/src/DotPulsar/Internal/ConsumerStream.cs
new file mode 100644
index 0000000..0dea1fd
--- /dev/null
+++ b/src/DotPulsar/Internal/ConsumerStream.cs
@@ -0,0 +1,151 @@
+﻿using DotPulsar.Exceptions;
+using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.Extensions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConsumerStream : IConsumerStream
+    {
+        private readonly ulong _id;
+        private readonly IDequeue<MessagePackage> _dequeue;
+        private readonly Connection _connection;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly IConsumerProxy _proxy;
+        private readonly CommandFlow _commandFlow;
+        private uint _sendWhenZero;
+        private bool _firstBatch;
+
+        public ConsumerStream(ulong id, uint messagePrefetchCount, IDequeue<MessagePackage> dequeue, Connection connection, IFaultStrategy faultStrategy, IConsumerProxy proxy)
+        {
+            _id = id;
+            _dequeue = dequeue;
+            _connection = connection;
+            _faultStrategy = faultStrategy;
+            _proxy = proxy;
+            _commandFlow = new CommandFlow { ConsumerId = id, MessagePermits = messagePrefetchCount };
+            _sendWhenZero = 0;
+            _firstBatch = true;
+        }
+
+        public async Task<Message> Receive(CancellationToken cancellationToken)
+        {
+            while (true)
+            {
+                if (_sendWhenZero == 0) //TODO should sending the flow command be handled on other thread and thereby not slow down the consumer?
+                {
+                    await _connection.Send(_commandFlow);
+
+                    if (_firstBatch)
+                    {
+                        _commandFlow.MessagePermits = (uint)Math.Ceiling(_commandFlow.MessagePermits * 0.5);
+                        _firstBatch = false;
+                    }
+
+                    _sendWhenZero = _commandFlow.MessagePermits;
+                }
+
+                var messagePackage = await _dequeue.Dequeue(cancellationToken);
+                _sendWhenZero--;
+
+                try
+                {
+                    return Serializer.Deserialize(messagePackage);
+                }
+                catch (ChecksumException)
+                {
+                    var ack = new CommandAck
+                    {
+                        Type = CommandAck.AckType.Individual,
+                        validation_error = CommandAck.ValidationError.ChecksumMismatch
+                    };
+                    ack.MessageIds.Add(messagePackage.Command.MessageId);
+                    await Send(ack);
+                }
+            }
+        }
+
+        public async Task Send(CommandAck command)
+        {
+            try
+            {
+                command.ConsumerId = _id;
+                await _connection.Send(command);
+            }
+            catch (Exception exception)
+            {
+                OnException(exception);
+                throw;
+            }
+        }
+
+        public async Task<CommandSuccess> Send(CommandUnsubscribe command)
+        {
+            try
+            {
+                command.ConsumerId = _id;
+                var response = await _connection.Send(command);
+                response.Expect(BaseCommand.Type.Success);
+                return response.Success;
+            }
+            catch (Exception exception)
+            {
+                OnException(exception);
+                throw;
+            }
+        }
+
+        public async Task<CommandSuccess> Send(CommandSeek command)
+        {
+            try
+            {
+                command.ConsumerId = _id;
+                var response = await _connection.Send(command);
+                response.Expect(BaseCommand.Type.Success);
+                return response.Success;
+            }
+            catch (Exception exception)
+            {
+                OnException(exception);
+                throw;
+            }
+        }
+
+        public async Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command)
+        {
+            try
+            {
+                command.ConsumerId = _id;
+                var response = await _connection.Send(command);
+                response.Expect(BaseCommand.Type.GetLastMessageIdResponse);
+                return response.GetLastMessageIdResponse;
+            }
+            catch (Exception exception)
+            {
+                OnException(exception);
+                throw;
+            }
+        }
+
+        public void Dispose()
+        {
+            try
+            {
+                _connection.Send(new CommandCloseConsumer { ConsumerId = _id }).Wait();
+            }
+            catch
+            {
+                // Ignore
+            }
+        }
+
+        private void OnException(Exception exception)
+        {
+            if (_faultStrategy.DetermineFaultAction(exception) == FaultAction.Relookup)
+                _proxy.Disconnected();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ConsumerStreamFactory.cs b/src/DotPulsar/Internal/ConsumerStreamFactory.cs
new file mode 100644
index 0000000..d8fd673
--- /dev/null
+++ b/src/DotPulsar/Internal/ConsumerStreamFactory.cs
@@ -0,0 +1,81 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ConsumerStreamFactory : IConsumerStreamFactory
+    {
+        private readonly ConnectionPool _connectionTool;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly CommandSubscribe _subscribe;
+        private readonly uint _messagePrefetchCount;
+
+        public ConsumerStreamFactory(ConnectionPool connectionManager, ConsumerOptions options, IFaultStrategy faultStrategy)
+        {
+            _connectionTool = connectionManager;
+            _faultStrategy = faultStrategy;
+            _messagePrefetchCount = options.MessagePrefetchCount;
+
+            _subscribe = new CommandSubscribe
+            {
+                ConsumerName = options.ConsumerName,
+                initialPosition = (CommandSubscribe.InitialPosition)options.InitialPosition,
+                PriorityLevel = options.PriorityLevel,
+                Subscription = options.SubscriptionName,
+                Topic = options.Topic,
+                Type = (CommandSubscribe.SubType)options.SubscriptionType
+            };
+        }
+
+        public ConsumerStreamFactory(ConnectionPool connectionManager, ReaderOptions options, IFaultStrategy faultStrategy)
+        {
+            _connectionTool = connectionManager;
+            _faultStrategy = faultStrategy;
+            _messagePrefetchCount = options.MessagePrefetchCount;
+
+            _subscribe = new CommandSubscribe
+            {
+                ConsumerName = options.ReaderName,
+                Durable = false,
+                StartMessageId = options.StartMessageId.Data,
+                Subscription = "Reader-" + Guid.NewGuid().ToString("N"),
+                Topic = options.Topic
+            };
+        }
+
+        public async Task<IConsumerStream> CreateStream(IConsumerProxy proxy, CancellationToken cancellationToken)
+        {
+            while (true)
+            {
+                try
+                {
+                    var connection = await _connectionTool.FindConnectionForTopic(_subscribe.Topic, cancellationToken);
+                    var response = await connection.Send(_subscribe, proxy);
+                    return new ConsumerStream(response.ConsumerId, _messagePrefetchCount, proxy, connection, _faultStrategy, proxy);
+                }
+                catch (OperationCanceledException)
+                {
+                    if (cancellationToken.IsCancellationRequested)
+                        throw;
+                    else
+                        await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                }
+                catch (Exception exception)
+                {
+                    switch (_faultStrategy.DetermineFaultAction(exception))
+                    {
+                        case FaultAction.Relookup:
+                        case FaultAction.Retry:
+                            await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                            continue;
+                    }
+
+                    throw;
+                }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Crc32C.cs b/src/DotPulsar/Internal/Crc32C.cs
new file mode 100644
index 0000000..eaee280
--- /dev/null
+++ b/src/DotPulsar/Internal/Crc32C.cs
@@ -0,0 +1,77 @@
+﻿using System.Buffers;
+
+namespace DotPulsar.Internal
+{
+    public static class Crc32C
+    {
+        private const uint Generator = 0x82F63B78u;
+
+        private static readonly uint[] Lookup;
+
+        static Crc32C()
+        {
+            Lookup = new uint[16 * 256];
+
+            for (uint i = 0; i < 256; i++)
+            {
+                var entry = i;
+                for (var j = 0; j < 16; j++)
+                {
+                    for (var k = 0; k < 8; k++)
+                    {
+                        entry = (entry & 1) == 1 ? Generator ^ (entry >> 1) : (entry >> 1);
+                    }
+                    Lookup[(j * 256) + i] = entry;
+                }
+            }
+        }
+
+        public static uint Calculate(ReadOnlySequence<byte> sequence)
+        {
+            var block = new uint[16];
+            var checksum = uint.MaxValue;
+            var remaningBytes = sequence.Length;
+            var readingBlock = remaningBytes >= 16;
+            var offset = 15;
+
+            foreach (var memory in sequence)
+            {
+                var span = memory.Span;
+                for (var i = 0; i < span.Length; ++i)
+                {
+                    var currentByte = span[i];
+
+                    if (!readingBlock)
+                    {
+                        checksum = Lookup[(byte)(checksum ^ currentByte)] ^ checksum >> 8;
+                        continue;
+                    }
+
+                    var offSetBase = offset * 256;
+
+                    if (offset > 11)
+                        block[offset] = Lookup[offSetBase + ((byte)(checksum >> (8 * (15 - offset))) ^ currentByte)];
+                    else
+                        block[offset] = Lookup[offSetBase + currentByte];
+
+                    --remaningBytes;
+
+                    if (offset == 0)
+                    {
+                        offset = 15;
+                        readingBlock = remaningBytes >= 16;
+                        checksum = 0;
+                        for (var j = 0; j < block.Length; ++j)
+                            checksum ^= block[j];
+                    }
+                    else
+                    {
+                        --offset;
+                    }
+                }
+            }
+
+            return checksum ^ uint.MaxValue;
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Exceptions/ConsumerNotFoundException.cs b/src/DotPulsar/Internal/Exceptions/ConsumerNotFoundException.cs
new file mode 100644
index 0000000..328cc70
--- /dev/null
+++ b/src/DotPulsar/Internal/Exceptions/ConsumerNotFoundException.cs
@@ -0,0 +1,9 @@
+﻿using DotPulsar.Exceptions;
+
+namespace DotPulsar.Internal.Exceptions
+{
+    public sealed class ConsumerNotFoundException : DotPulsarException
+    {
+        public ConsumerNotFoundException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Internal/Exceptions/ServiceNotReadyException.cs b/src/DotPulsar/Internal/Exceptions/ServiceNotReadyException.cs
new file mode 100644
index 0000000..ca43151
--- /dev/null
+++ b/src/DotPulsar/Internal/Exceptions/ServiceNotReadyException.cs
@@ -0,0 +1,9 @@
+﻿using DotPulsar.Exceptions;
+
+namespace DotPulsar.Internal.Exceptions
+{
+    public sealed class ServiceNotReadyException : DotPulsarException
+    {
+        public ServiceNotReadyException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Internal/Exceptions/StreamNotReadyException.cs b/src/DotPulsar/Internal/Exceptions/StreamNotReadyException.cs
new file mode 100644
index 0000000..5618e19
--- /dev/null
+++ b/src/DotPulsar/Internal/Exceptions/StreamNotReadyException.cs
@@ -0,0 +1,9 @@
+﻿using DotPulsar.Exceptions;
+
+namespace DotPulsar.Internal.Exceptions
+{
+    public sealed class StreamNotReadyException : DotPulsarException
+    {
+        public StreamNotReadyException() : base("The service is not ready yet") { }
+    }
+}
diff --git a/src/DotPulsar/Internal/Exceptions/TooManyRequestsException.cs b/src/DotPulsar/Internal/Exceptions/TooManyRequestsException.cs
new file mode 100644
index 0000000..d44620f
--- /dev/null
+++ b/src/DotPulsar/Internal/Exceptions/TooManyRequestsException.cs
@@ -0,0 +1,9 @@
+﻿using DotPulsar.Exceptions;
+
+namespace DotPulsar.Internal.Exceptions
+{
+    public sealed class TooManyRequestsException : DotPulsarException
+    {
+        public TooManyRequestsException(string message) : base(message) { }
+    }
+}
diff --git a/src/DotPulsar/Internal/Exceptions/UnexpectedResponseException.cs b/src/DotPulsar/Internal/Exceptions/UnexpectedResponseException.cs
new file mode 100644
index 0000000..d783deb
--- /dev/null
+++ b/src/DotPulsar/Internal/Exceptions/UnexpectedResponseException.cs
@@ -0,0 +1,11 @@
+﻿using DotPulsar.Exceptions;
+using System;
+
+namespace DotPulsar.Internal.Exceptions
+{
+    public sealed class UnexpectedResponseException : DotPulsarException
+    {
+        public UnexpectedResponseException(string message) : base(message) { }
+        public UnexpectedResponseException(string message, Exception innerException) : base(message, innerException) { }
+    }
+}
diff --git a/src/DotPulsar/Internal/Executor.cs b/src/DotPulsar/Internal/Executor.cs
new file mode 100644
index 0000000..1f5b5b6
--- /dev/null
+++ b/src/DotPulsar/Internal/Executor.cs
@@ -0,0 +1,102 @@
+﻿using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Executor : IDisposable
+    {
+        private readonly AsyncLock _lock;
+        private readonly Func<Exception, CancellationToken, Task> _onException;
+
+        public Executor(Func<Exception, CancellationToken, Task> onException)
+        {
+            _lock = new AsyncLock();
+            _onException = onException;
+        }
+
+        public void Dispose() => _lock.Dispose();
+
+        public async Task Execute(Action action, CancellationToken cancellationToken)
+        {
+            using (await _lock.Lock(cancellationToken))
+            {
+                while (true)
+                {
+                    try
+                    {
+                        action();
+                        return;
+                    }
+                    catch (Exception exception)
+                    {
+                        await _onException(exception, cancellationToken);
+                    }
+
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+            }
+        }
+
+        public async Task Execute(Func<Task> func, CancellationToken cancellationToken)
+        {
+            using (await _lock.Lock(cancellationToken))
+            {
+                while (true)
+                {
+                    try
+                    {
+                        await func();
+                        return;
+                    }
+                    catch (Exception exception)
+                    {
+                        await _onException(exception, cancellationToken);
+                    }
+
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+            }
+        }
+
+        public async Task<TResult> Execute<TResult>(Func<TResult> func, CancellationToken cancellationToken)
+        {
+            using (await _lock.Lock(cancellationToken))
+            {
+                while (true)
+                {
+                    try
+                    {
+                        return func();
+                    }
+                    catch (Exception exception)
+                    {
+                        await _onException(exception, cancellationToken);
+                    }
+
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+            }
+        }
+
+        public async Task<TResult> Execute<TResult>(Func<Task<TResult>> func, CancellationToken cancellationToken)
+        {
+            using (await _lock.Lock(cancellationToken))
+            {
+                while (true)
+                {
+                    try
+                    {
+                        return await func();
+                    }
+                    catch (Exception exception)
+                    {
+                        await _onException(exception, cancellationToken);
+                    }
+
+                    cancellationToken.ThrowIfCancellationRequested();
+                }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Extensions/CommandExtensions.cs b/src/DotPulsar/Internal/Extensions/CommandExtensions.cs
new file mode 100644
index 0000000..29d743c
--- /dev/null
+++ b/src/DotPulsar/Internal/Extensions/CommandExtensions.cs
@@ -0,0 +1,193 @@
+﻿using DotPulsar.Exceptions;
+using DotPulsar.Internal.Exceptions;
+using DotPulsar.Internal.PulsarApi;
+
+namespace DotPulsar.Internal.Extensions
+{
+    public static class CommandExtensions
+    {
+        public static void Expect(this BaseCommand command, params BaseCommand.Type[] types)
+        {
+            var actual = command.CommandType;
+
+            foreach (var type in types)
+            {
+                if (type == actual)
+                    return;
+            }
+
+            switch (actual)
+            {
+                case BaseCommand.Type.Error:
+                    command.Error.Throw();
+                    return;
+                case BaseCommand.Type.SendError:
+                    command.SendError.Throw();
+                    return;
+            }
+
+            throw new UnexpectedResponseException($"Expected '{string.Join(",", types)}' but got '{actual}'");
+        }
+
+        public static void Throw(this CommandSendError command) => Throw(command.Error, command.Message);
+
+        public static void Throw(this CommandLookupTopicResponse command) => Throw(command.Error, command.Message);
+
+        public static void Throw(this CommandError error) => Throw(error.Error, error.Message);
+
+        private static void Throw(ServerError error, string message)
+        {
+            switch (error)
+            {
+                case ServerError.AuthenticationError: throw new AuthenticationException(message);
+                case ServerError.AuthorizationError: throw new AuthorizationException(message);
+                case ServerError.ChecksumError: throw new ChecksumException(message);
+                case ServerError.ConsumerAssignError: throw new ConsumerAssignException(message);
+                case ServerError.ConsumerBusy: throw new ConsumerBusyException(message);
+                case ServerError.ConsumerNotFound: throw new ConsumerNotFoundException(message);
+                case ServerError.IncompatibleSchema: throw new IncompatibleSchemaException(message);
+                case ServerError.InvalidTopicName: throw new InvalidTopicNameException(message);
+                case ServerError.MetadataError: throw new MetadataException(message);
+                case ServerError.PersistenceError: throw new PersistenceException(message);
+                case ServerError.ProducerBlockedQuotaExceededError:
+                case ServerError.ProducerBlockedQuotaExceededException:
+                    throw new ProducerBlockedQuotaExceededException(message + ". Error code: " + error);
+                case ServerError.ProducerBusy: throw new ProducerBusyException(message);
+                case ServerError.ServiceNotReady: throw new ServiceNotReadyException(message);
+                case ServerError.SubscriptionNotFound: throw new SubscriptionNotFoundException(message);
+                case ServerError.TooManyRequests: throw new TooManyRequestsException(message);
+                case ServerError.TopicNotFound: throw new TopicNotFoundException(message);
+                case ServerError.TopicTerminatedError: throw new TopicTerminatedException(message);
+                case ServerError.UnsupportedVersionError: throw new UnsupportedVersionException(message);
+                case ServerError.UnknownError:
+                default: throw new UnknownException(message + ". Error code: " + error);
+            }
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandAck command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Ack,
+                Ack = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandConnect command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Connect,
+                Connect = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandPing command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Ping,
+                Ping = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandPong command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Pong,
+                Pong = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandProducer command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Producer,
+                Producer = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandGetLastMessageId command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.GetLastMessageId,
+                GetLastMessageId = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandUnsubscribe command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Unsubscribe,
+                Unsubscribe = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandSubscribe command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Subscribe,
+                Subscribe = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandLookupTopic command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Lookup,
+                LookupTopic = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandSend command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Send,
+                Send = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandFlow command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Flow,
+                Flow = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandCloseProducer command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.CloseProducer,
+                CloseProducer = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandCloseConsumer command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.CloseConsumer,
+                CloseConsumer = command
+            };
+        }
+
+        public static BaseCommand AsBaseCommand(this CommandSeek command)
+        {
+            return new BaseCommand
+            {
+                CommandType = BaseCommand.Type.Seek,
+                Seek = command
+            };
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs b/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs
new file mode 100644
index 0000000..e477974
--- /dev/null
+++ b/src/DotPulsar/Internal/Extensions/ReadOnlySequenceExtensions.cs
@@ -0,0 +1,82 @@
+﻿using System;
+using System.Buffers;
+
+namespace DotPulsar.Internal.Extensions
+{
+    public static class ReadOnlySequenceExtensions
+    {
+        public static bool StartsWith<T>(this ReadOnlySequence<T> sequence, ReadOnlyMemory<T> target) where T : IEquatable<T>
+        {
+            if (target.Length > sequence.Length)
+                return false;
+
+            var targetIndex = 0;
+            var targetSpan = target.Span;
+
+            foreach (var memory in sequence)
+            {
+                var span = memory.Span;
+                for (var i = 0; i < span.Length; ++i)
+                {
+                    if (!span[i].Equals(targetSpan[targetIndex]))
+                        return false;
+
+                    ++targetIndex;
+                    if (targetIndex == targetSpan.Length)
+                        return true;
+                }
+            }
+
+            return false;
+        }
+
+        public static uint ReadUInt32(this ReadOnlySequence<byte> sequence, int start, bool isBigEndian)
+        {
+            if (sequence.Length < (4 + start))
+                throw new ArgumentOutOfRangeException(nameof(start), start, "Sequence must be at least 4 bytes long from 'start' to end");
+
+            var reverse = isBigEndian != BitConverter.IsLittleEndian;
+            var union = new UIntUnion();
+            var read = 0;
+
+            foreach (var memory in sequence)
+            {
+                if (start > memory.Length)
+                {
+                    start -= memory.Length;
+                    continue;
+                }
+
+                var span = memory.Span;
+                for (var i = start; i < span.Length; ++i, ++read)
+                {
+                    switch (read)
+                    {
+                        case 0:
+                            if (reverse) union.B0 = span[i];
+                            else union.B3 = span[i];
+                            continue;
+                        case 1:
+                            if (reverse) union.B1 = span[i];
+                            else union.B2 = span[i];
+                            continue;
+                        case 2:
+                            if (reverse) union.B2 = span[i];
+                            else union.B1 = span[i];
+                            continue;
+                        case 3:
+                            if (reverse) union.B3 = span[i];
+                            else union.B0 = span[i];
+                            break;
+                    }
+                }
+
+                if (read == 3)
+                    break;
+                start = 0;
+            }
+
+            return union.UInt;
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/FaultAction.cs b/src/DotPulsar/Internal/FaultAction.cs
new file mode 100644
index 0000000..263567a
--- /dev/null
+++ b/src/DotPulsar/Internal/FaultAction.cs
@@ -0,0 +1,9 @@
+﻿namespace DotPulsar.Internal
+{
+    public enum FaultAction : byte
+    {
+        Retry,
+        Relookup,
+        Fault
+    }
+}
diff --git a/src/DotPulsar/Internal/FaultStrategy.cs b/src/DotPulsar/Internal/FaultStrategy.cs
new file mode 100644
index 0000000..cfd91a0
--- /dev/null
+++ b/src/DotPulsar/Internal/FaultStrategy.cs
@@ -0,0 +1,30 @@
+﻿using DotPulsar.Exceptions;
+using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.Exceptions;
+using System;
+
+namespace DotPulsar.Internal
+{
+    public sealed class FaultStrategy : IFaultStrategy
+    {
+        public FaultStrategy(int timeToWaitInMilliseconds)
+        {
+            TimeToWait = TimeSpan.FromMilliseconds(timeToWaitInMilliseconds);
+        }
+
+        public TimeSpan TimeToWait { get; }
+
+        public FaultAction DetermineFaultAction(Exception exception)
+        {
+            switch (exception)
+            {
+                case TooManyRequestsException _: return FaultAction.Retry;
+                case StreamNotReadyException _: return FaultAction.Relookup;
+                case ServiceNotReadyException _: return FaultAction.Relookup;
+                case DotPulsarException _: return FaultAction.Fault;
+            }
+
+            return FaultAction.Relookup;
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/IdLookup.cs b/src/DotPulsar/Internal/IdLookup.cs
new file mode 100644
index 0000000..433d9e2
--- /dev/null
+++ b/src/DotPulsar/Internal/IdLookup.cs
@@ -0,0 +1,59 @@
+﻿using System.Collections.Generic;
+
+namespace DotPulsar.Internal
+{
+    public sealed class IdLookup<T> where T : class
+    {
+        private T[] _items;
+
+        public IdLookup() => _items = new T[0];
+
+        public bool IsEmpty()
+        {
+            for (var i = 0; i < _items.Length; ++i)
+            {
+                if (_items[i] != null)
+                    return false;
+            }
+
+            return true;
+        }
+
+        public ulong[] AllIds()
+        {
+            var activeIds = new List<ulong>();
+            for (var i = 0; i < _items.Length; ++i)
+            {
+                if (_items[i] != null)
+                    activeIds.Add((ulong)i);
+            }
+            return activeIds.ToArray();
+        }
+
+        public ulong Add(T item)
+        {
+            for (int i = 0; i < _items.Length; ++i)
+            {
+                if (_items[i] != null)
+                    continue;
+
+                _items[i] = item;
+                return (ulong)i;
+            }
+
+            var newArray = new T[_items.Length + 1];
+            _items.CopyTo(newArray, 0);
+            var id = newArray.Length - 1;
+            newArray[id] = item;
+            _items = newArray;
+            return (ulong)id;
+        }
+
+        public void Remove(ulong id) => _items[(int)id] = null;
+
+        public T this[ulong id]
+        {
+            get => _items[(int)id];
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/MessageBuilder.cs b/src/DotPulsar/Internal/MessageBuilder.cs
new file mode 100644
index 0000000..394a3f6
--- /dev/null
+++ b/src/DotPulsar/Internal/MessageBuilder.cs
@@ -0,0 +1,39 @@
+﻿using DotPulsar.Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class MessageBuilder : IMessageBuilder
+    {
+        private readonly IProducer _producer;
+        private readonly MessageMetadata _metadata;
+
+        public MessageBuilder(IProducer producer)
+        {
+            _producer = producer;
+            _metadata = new MessageMetadata();
+        }
+
+        public IMessageBuilder EventTime(ulong eventTime)
+        {
+            _metadata.EventTime = eventTime;
+            return this;
+        }
+
+        public IMessageBuilder Property(string key, string value)
+        {
+            _metadata[key] = value;
+            return this;
+        }
+
+        public IMessageBuilder SequenceId(ulong sequenceId)
+        {
+            _metadata.SequenceId = sequenceId;
+            return this;
+        }
+
+        public async Task<MessageId> Send(ReadOnlyMemory<byte> data, CancellationToken cancellationToken) => await _producer.Send(_metadata, data, cancellationToken);
+    }
+}
diff --git a/src/DotPulsar/Internal/MessagePackage.cs b/src/DotPulsar/Internal/MessagePackage.cs
new file mode 100644
index 0000000..2aaf738
--- /dev/null
+++ b/src/DotPulsar/Internal/MessagePackage.cs
@@ -0,0 +1,17 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System.Buffers;
+
+namespace DotPulsar.Internal
+{
+    public sealed class MessagePackage
+    {
+        public MessagePackage(CommandMessage command, ReadOnlySequence<byte> data)
+        {
+            Command = command;
+            Data = data;
+        }
+
+        public CommandMessage Command { get; }
+        public ReadOnlySequence<byte> Data { get; }
+    }
+}
diff --git a/src/DotPulsar/Internal/NotReadyStream.cs b/src/DotPulsar/Internal/NotReadyStream.cs
new file mode 100644
index 0000000..0b4f888
--- /dev/null
+++ b/src/DotPulsar/Internal/NotReadyStream.cs
@@ -0,0 +1,28 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.Exceptions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class NotReadyStream : IConsumerStream, IProducerStream
+    {
+        public void Dispose() { }
+
+        public Task<Message> Receive(CancellationToken cancellationToken) => throw GetException();
+
+        public Task Send(CommandAck command) => throw GetException();
+
+        public Task<CommandSuccess> Send(CommandUnsubscribe command) => throw GetException();
+
+        public Task<CommandSuccess> Send(CommandSeek command) => throw GetException();
+
+        public Task<CommandGetLastMessageIdResponse> Send(CommandGetLastMessageId command) => throw GetException();
+
+        public Task<CommandSendReceipt> Send(PulsarApi.MessageMetadata metadata, ReadOnlyMemory<byte> payload) => throw GetException();
+
+        private Exception GetException() => new StreamNotReadyException();
+    }
+}
diff --git a/src/DotPulsar/Internal/PingPongHandler.cs b/src/DotPulsar/Internal/PingPongHandler.cs
new file mode 100644
index 0000000..e2367b4
--- /dev/null
+++ b/src/DotPulsar/Internal/PingPongHandler.cs
@@ -0,0 +1,21 @@
+﻿using DotPulsar.Internal.PulsarApi;
+
+namespace DotPulsar.Internal
+{
+    public sealed class PingPongHandler
+    {
+        private readonly Connection _connection;
+        private readonly CommandPong _pong;
+
+        public PingPongHandler(Connection connection)
+        {
+            _connection = connection;
+            _pong = new CommandPong();
+        }
+
+        public void Incoming(CommandPing ping)
+        {
+            _ = _connection.Send(_pong);
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Producer.cs b/src/DotPulsar/Internal/Producer.cs
new file mode 100644
index 0000000..d37d351
--- /dev/null
+++ b/src/DotPulsar/Internal/Producer.cs
@@ -0,0 +1,116 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Exceptions;
+using DotPulsar.Internal.Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Producer : IProducer
+    {
+        private readonly Executor _executor;
+        private readonly IProducerStreamFactory _streamFactory;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly StateManager<ProducerState> _stateManager;
+        private readonly CancellationTokenSource _connectTokenSource;
+        private readonly Task _connectTask;
+        private Action _throwIfClosedOrFaulted;
+        private IProducerStream Stream { get; set; }
+
+        public Producer(IProducerStreamFactory streamFactory, IFaultStrategy faultStrategy)
+        {
+            _executor = new Executor(ExecutorOnException);
+            _streamFactory = streamFactory;
+            _faultStrategy = faultStrategy;
+            _stateManager = new StateManager<ProducerState>(ProducerState.Disconnected, ProducerState.Closed, ProducerState.Faulted);
+            _connectTokenSource = new CancellationTokenSource();
+            Stream = new NotReadyStream();
+            _connectTask = Connect(_connectTokenSource.Token);
+            _throwIfClosedOrFaulted = () => { };
+        }
+
+        public async Task<ProducerState> StateChangedTo(ProducerState state, CancellationToken cancellationToken) => await _stateManager.StateChangedTo(state, cancellationToken);
+        public async Task<ProducerState> StateChangedFrom(ProducerState state, CancellationToken cancellationToken) => await _stateManager.StateChangedFrom(state, cancellationToken);
+        public bool IsFinalState() => _stateManager.IsFinalState();
+        public bool IsFinalState(ProducerState state) => _stateManager.IsFinalState(state);
+
+        public void Dispose()
+        {
+            _executor.Dispose();
+            _connectTokenSource.Cancel();
+            _connectTask.Wait();
+        }
+
+        public async Task<MessageId> Send(ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
+            => await Send(new MessageMetadata(), data, cancellationToken);
+
+        public async Task<MessageId> Send(MessageMetadata metadata, ReadOnlyMemory<byte> data, CancellationToken cancellationToken)
+            => await Send(metadata.Metadata, data, cancellationToken);
+
+        private async Task<MessageId> Send(PulsarApi.MessageMetadata metadata, ReadOnlyMemory<byte> payload, CancellationToken cancellationToken)
+        {
+            var response = await _executor.Execute(() => Stream.Send(metadata, payload), cancellationToken);
+            return new MessageId(response.MessageId);
+        }
+
+        private async Task ExecutorOnException(Exception exception, CancellationToken cancellationToken)
+        {
+            _throwIfClosedOrFaulted();
+
+            switch (_faultStrategy.DetermineFaultAction(exception))
+            {
+                case FaultAction.Retry:
+                    await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                    break;
+                case FaultAction.Relookup:
+                    await _stateManager.StateChangedTo(ProducerState.Connected, cancellationToken);
+                    break;
+                case FaultAction.Fault:
+                    HasFaulted(exception);
+                    break;
+            }
+
+            _throwIfClosedOrFaulted();
+        }
+
+        private void HasFaulted(Exception exception)
+        {
+            _throwIfClosedOrFaulted = () => throw exception;
+            _stateManager.SetState(ProducerState.Faulted);
+        }
+
+        private void HasClosed()
+        {
+            _throwIfClosedOrFaulted = () => throw new ProducerClosedException();
+            _stateManager.SetState(ProducerState.Closed);
+        }
+
+        private async Task Connect(CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (true)
+                {
+                    var proxy = new ProducerProxy(_stateManager);
+
+                    using (Stream = await _streamFactory.CreateStream(proxy, cancellationToken))
+                    {
+                        proxy.Connected();
+                        await _stateManager.StateChangedFrom(ProducerState.Connected, cancellationToken);
+                        if (_stateManager.IsFinalState())
+                            return;
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                HasClosed();
+            }
+            catch (Exception exception)
+            {
+                HasFaulted(exception);
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerBuilder.cs b/src/DotPulsar/Internal/ProducerBuilder.cs
new file mode 100644
index 0000000..654086c
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerBuilder.cs
@@ -0,0 +1,36 @@
+﻿using DotPulsar.Abstractions;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ProducerBuilder : IProducerBuilder
+    {
+        private readonly IPulsarClient _pulsarClient;
+        private readonly ProducerOptions _options;
+
+        public ProducerBuilder(IPulsarClient pulsarClient)
+        {
+            _pulsarClient = pulsarClient;
+            _options = new ProducerOptions();
+        }
+
+        public IProducerBuilder ProducerName(string name)
+        {
+            _options.ProducerName = name;
+            return this;
+        }
+
+        public IProducerBuilder InitialSequenceId(ulong initialSequenceId)
+        {
+            _options.InitialSequenceId = initialSequenceId;
+            return this;
+        }
+
+        public IProducerBuilder Topic(string topic)
+        {
+            _options.Topic = topic;
+            return this;
+        }
+
+        public IProducer Create() => _pulsarClient.CreateProducer(_options);
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerManager.cs b/src/DotPulsar/Internal/ProducerManager.cs
new file mode 100644
index 0000000..2bd6494
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerManager.cs
@@ -0,0 +1,37 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ProducerManager : IDisposable
+    {
+        private readonly IdLookup<IProducerProxy> _proxies;
+
+        public ProducerManager() => _proxies = new IdLookup<IProducerProxy>();
+
+        public bool HasProducers => !_proxies.IsEmpty();
+
+        public void Outgoing(CommandProducer producer, IProducerProxy proxy) => producer.ProducerId = _proxies.Add(proxy);
+
+        public void Remove(ulong producerId) => _proxies.Remove(producerId);
+
+        public void Incoming(CommandCloseProducer command) => RemoveProducer(command.ProducerId);
+
+        public void Dispose()
+        {
+            foreach (var id in _proxies.AllIds())
+            {
+                RemoveProducer(id);
+            }
+        }
+
+        private void RemoveProducer(ulong producerId)
+        {
+            var stateManager = _proxies[producerId];
+            if (stateManager == null) return;
+            stateManager.Disconnected();
+            _proxies.Remove(producerId);
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerProxy.cs b/src/DotPulsar/Internal/ProducerProxy.cs
new file mode 100644
index 0000000..f5711e5
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerProxy.cs
@@ -0,0 +1,39 @@
+﻿using DotPulsar.Internal.Abstractions;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ProducerProxy : IProducerProxy
+    {
+        private readonly object _lock;
+        private readonly StateManager<ProducerState> _stateManager;
+        private bool _hasDisconnected;
+
+        public ProducerProxy(StateManager<ProducerState> stateManager)
+        {
+            _lock = new object();
+            _stateManager = stateManager;
+            _hasDisconnected = false;
+        }
+
+        public void Connected()
+        {
+            lock (_lock)
+            {
+                if (!_hasDisconnected)
+                    _stateManager.SetState(ProducerState.Connected);
+            }
+        }
+
+        public void Disconnected()
+        {
+            lock (_lock)
+            {
+                if (_hasDisconnected)
+                    return;
+
+                _stateManager.SetState(ProducerState.Disconnected);
+                _hasDisconnected = true;
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerResponse.cs b/src/DotPulsar/Internal/ProducerResponse.cs
new file mode 100644
index 0000000..458d186
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerResponse.cs
@@ -0,0 +1,14 @@
+﻿namespace DotPulsar.Internal
+{
+    public sealed class ProducerResponse
+    {
+        public ProducerResponse(ulong producerId, string producerName)
+        {
+            ProducerId = producerId;
+            ProducerName = producerName;
+        }
+
+        public ulong ProducerId { get; }
+        public string ProducerName { get; }
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerStream.cs b/src/DotPulsar/Internal/ProducerStream.cs
new file mode 100644
index 0000000..7b296e0
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerStream.cs
@@ -0,0 +1,69 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.Extensions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ProducerStream : IProducerStream
+    {
+        private readonly SendPackage _cachedSendPackage;
+        private readonly ulong _id;
+        private readonly string _name;
+        private readonly SequenceId _sequenceId;
+        private readonly Connection _connection;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly IProducerProxy _proxy;
+
+        public ProducerStream(ulong id, string name, SequenceId sequenceId, Connection connection, IFaultStrategy faultStrategy, IProducerProxy proxy)
+        {
+            _cachedSendPackage = new SendPackage(new CommandSend { ProducerId = id, NumMessages = 1 });
+            _id = id;
+            _name = name;
+            _sequenceId = sequenceId;
+            _connection = connection;
+            _faultStrategy = faultStrategy;
+            _proxy = proxy;
+        }
+
+        public void Dispose()
+        {
+            try
+            {
+                _connection.Send(new CommandCloseProducer { ProducerId = _id }).Wait();
+            }
+            catch
+            {
+                // Ignore
+            }
+        }
+
+        public async Task<CommandSendReceipt> Send(PulsarApi.MessageMetadata metadata, ReadOnlyMemory<byte> payload)
+        {
+            try
+            {
+                metadata.PublishTime = (ulong)DateTimeOffset.UtcNow.ToUnixTimeMilliseconds();
+                metadata.ProducerName = _name;
+
+                if (metadata.SequenceId == 0)
+                    metadata.SequenceId = _sequenceId.Current;
+
+                _cachedSendPackage.Command.SequenceId = metadata.SequenceId;
+                _cachedSendPackage.Metadata = metadata;
+                _cachedSendPackage.Payload = payload;
+
+                var response = await _connection.Send(_cachedSendPackage);
+                response.Expect(BaseCommand.Type.SendReceipt); //TODO find out if we should increment on SendError
+                _sequenceId.Increment();
+                return response.SendReceipt;
+            }
+            catch (Exception exception)
+            {
+                if (_faultStrategy.DetermineFaultAction(exception) == FaultAction.Relookup)
+                    _proxy.Disconnected();
+                throw;
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ProducerStreamFactory.cs b/src/DotPulsar/Internal/ProducerStreamFactory.cs
new file mode 100644
index 0000000..5c3f138
--- /dev/null
+++ b/src/DotPulsar/Internal/ProducerStreamFactory.cs
@@ -0,0 +1,62 @@
+﻿using DotPulsar.Internal.Abstractions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ProducerStreamFactory : IProducerStreamFactory
+    {
+        private readonly ConnectionPool _connectionTool;
+        private readonly ProducerOptions _options;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly SequenceId _sequenceId;
+
+        public ProducerStreamFactory(ConnectionPool connectionManager, ProducerOptions options, IFaultStrategy faultStrategy)
+        {
+            _connectionTool = connectionManager;
+            _options = options;
+            _faultStrategy = faultStrategy;
+            _sequenceId = new SequenceId(options.InitialSequenceId);
+        }
+
+        public async Task<IProducerStream> CreateStream(IProducerProxy proxy, CancellationToken cancellationToken)
+        {
+            var commandProducer = new CommandProducer
+            {
+                ProducerName = _options.ProducerName,
+                Topic = _options.Topic
+            };
+
+            while (true)
+            {
+                try
+                {
+                    var connection = await _connectionTool.FindConnectionForTopic(_options.Topic, cancellationToken);
+                    var response = await connection.Send(commandProducer, proxy);
+                    return new ProducerStream(response.ProducerId, response.ProducerName, _sequenceId, connection, _faultStrategy, proxy);
+                }
+                catch (OperationCanceledException)
+                {
+                    if (cancellationToken.IsCancellationRequested)
+                        throw;
+                    else
+                        await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                }
+                catch (Exception exception)
+                {
+                    switch (_faultStrategy.DetermineFaultAction(exception))
+                    {
+                        case FaultAction.Relookup:
+                        case FaultAction.Retry:
+                            await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                            continue;
+                    }
+
+                    throw;
+                }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/PulsarApi/GeneratedCode.cs b/src/DotPulsar/Internal/PulsarApi/GeneratedCode.cs
new file mode 100644
index 0000000..7560c72
--- /dev/null
+++ b/src/DotPulsar/Internal/PulsarApi/GeneratedCode.cs
@@ -0,0 +1,1953 @@
+﻿namespace DotPulsar.Internal.PulsarApi
+{
+    // This file was generated by a tool; you should avoid making direct changes.
+    // Consider using 'partial classes' to extend these types
+    // Input: PulsarApi.proto (Last edit: ba24d73)
+
+#pragma warning disable CS1591, CS0612, CS3021, IDE1006
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class Schema : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"name", IsRequired = true)]
+        public string Name { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"schema_data", IsRequired = true)]
+        public byte[] SchemaData { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, IsRequired = true)]
+        public SchemaType Type { get; set; }
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"properties")]
+        public global::System.Collections.Generic.List<KeyValue> Properties { get; } = new global::System.Collections.Generic.List<KeyValue>();
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum SchemaType
+        {
+            None = 0,
+            String = 1,
+            Json = 2,
+            Protobuf = 3,
+            Avro = 4,
+            Bool = 5,
+            Int8 = 6,
+            Int16 = 7,
+            Int32 = 8,
+            Int64 = 9,
+            Float = 10,
+            Double = 11,
+            Date = 12,
+            Time = 13,
+            Timestamp = 14,
+            KeyValue = 15,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class MessageIdData : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, IsRequired = true)]
+        public ulong LedgerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, IsRequired = true)]
+        public ulong EntryId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"partition")]
+        [global::System.ComponentModel.DefaultValue(-1)]
+        public int Partition
+        {
+            get { return __pbn__Partition ?? -1; }
+            set { __pbn__Partition = value; }
+        }
+        public bool ShouldSerializePartition() => __pbn__Partition != null;
+        public void ResetPartition() => __pbn__Partition = null;
+        private int? __pbn__Partition;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"batch_index")]
+        [global::System.ComponentModel.DefaultValue(-1)]
+        public int BatchIndex
+        {
+            get { return __pbn__BatchIndex ?? -1; }
+            set { __pbn__BatchIndex = value; }
+        }
+        public bool ShouldSerializeBatchIndex() => __pbn__BatchIndex != null;
+        public void ResetBatchIndex() => __pbn__BatchIndex = null;
+        private int? __pbn__BatchIndex;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class KeyValue : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"key", IsRequired = true)]
+        public string Key { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"value", IsRequired = true)]
+        public string Value { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class KeyLongValue : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"key", IsRequired = true)]
+        public string Key { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"value", IsRequired = true)]
+        public ulong Value { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class EncryptionKeys : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"key", IsRequired = true)]
+        public string Key { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"value", IsRequired = true)]
+        public byte[] Value { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"metadata")]
+        public global::System.Collections.Generic.List<KeyValue> Metadatas { get; } = new global::System.Collections.Generic.List<KeyValue>();
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class MessageMetadata : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"producer_name", IsRequired = true)]
+        public string ProducerName { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"sequence_id", IsRequired = true)]
+        public ulong SequenceId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"publish_time", IsRequired = true)]
+        public ulong PublishTime { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"properties")]
+        public global::System.Collections.Generic.List<KeyValue> Properties { get; } = new global::System.Collections.Generic.List<KeyValue>();
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"replicated_from")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ReplicatedFrom
+        {
+            get { return __pbn__ReplicatedFrom ?? ""; }
+            set { __pbn__ReplicatedFrom = value; }
+        }
+        public bool ShouldSerializeReplicatedFrom() => __pbn__ReplicatedFrom != null;
+        public void ResetReplicatedFrom() => __pbn__ReplicatedFrom = null;
+        private string __pbn__ReplicatedFrom;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"partition_key")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string PartitionKey
+        {
+            get { return __pbn__PartitionKey ?? ""; }
+            set { __pbn__PartitionKey = value; }
+        }
+        public bool ShouldSerializePartitionKey() => __pbn__PartitionKey != null;
+        public void ResetPartitionKey() => __pbn__PartitionKey = null;
+        private string __pbn__PartitionKey;
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"replicate_to")]
+        public global::System.Collections.Generic.List<string> ReplicateToes { get; } = new global::System.Collections.Generic.List<string>();
+
+        [global::ProtoBuf.ProtoMember(8, Name = @"compression")]
+        [global::System.ComponentModel.DefaultValue(CompressionType.None)]
+        public CompressionType Compression
+        {
+            get { return __pbn__Compression ?? CompressionType.None; }
+            set { __pbn__Compression = value; }
+        }
+        public bool ShouldSerializeCompression() => __pbn__Compression != null;
+        public void ResetCompression() => __pbn__Compression = null;
+        private CompressionType? __pbn__Compression;
+
+        [global::ProtoBuf.ProtoMember(9, Name = @"uncompressed_size")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public uint UncompressedSize
+        {
+            get { return __pbn__UncompressedSize ?? 0; }
+            set { __pbn__UncompressedSize = value; }
+        }
+        public bool ShouldSerializeUncompressedSize() => __pbn__UncompressedSize != null;
+        public void ResetUncompressedSize() => __pbn__UncompressedSize = null;
+        private uint? __pbn__UncompressedSize;
+
+        [global::ProtoBuf.ProtoMember(11, Name = @"num_messages_in_batch")]
+        [global::System.ComponentModel.DefaultValue(1)]
+        public int NumMessagesInBatch
+        {
+            get { return __pbn__NumMessagesInBatch ?? 1; }
+            set { __pbn__NumMessagesInBatch = value; }
+        }
+        public bool ShouldSerializeNumMessagesInBatch() => __pbn__NumMessagesInBatch != null;
+        public void ResetNumMessagesInBatch() => __pbn__NumMessagesInBatch = null;
+        private int? __pbn__NumMessagesInBatch;
+
+        [global::ProtoBuf.ProtoMember(12, Name = @"event_time")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public ulong EventTime
+        {
+            get { return __pbn__EventTime ?? 0; }
+            set { __pbn__EventTime = value; }
+        }
+        public bool ShouldSerializeEventTime() => __pbn__EventTime != null;
+        public void ResetEventTime() => __pbn__EventTime = null;
+        private ulong? __pbn__EventTime;
+
+        [global::ProtoBuf.ProtoMember(13, Name = @"encryption_keys")]
+        public global::System.Collections.Generic.List<EncryptionKeys> EncryptionKeys { get; } = new global::System.Collections.Generic.List<EncryptionKeys>();
+
+        [global::ProtoBuf.ProtoMember(14, Name = @"encryption_algo")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string EncryptionAlgo
+        {
+            get { return __pbn__EncryptionAlgo ?? ""; }
+            set { __pbn__EncryptionAlgo = value; }
+        }
+        public bool ShouldSerializeEncryptionAlgo() => __pbn__EncryptionAlgo != null;
+        public void ResetEncryptionAlgo() => __pbn__EncryptionAlgo = null;
+        private string __pbn__EncryptionAlgo;
+
+        [global::ProtoBuf.ProtoMember(15, Name = @"encryption_param")]
+        public byte[] EncryptionParam
+        {
+            get { return __pbn__EncryptionParam; }
+            set { __pbn__EncryptionParam = value; }
+        }
+        public bool ShouldSerializeEncryptionParam() => __pbn__EncryptionParam != null;
+        public void ResetEncryptionParam() => __pbn__EncryptionParam = null;
+        private byte[] __pbn__EncryptionParam;
+
+        [global::ProtoBuf.ProtoMember(16, Name = @"schema_version")]
+        public byte[] SchemaVersion
+        {
+            get { return __pbn__SchemaVersion; }
+            set { __pbn__SchemaVersion = value; }
+        }
+        public bool ShouldSerializeSchemaVersion() => __pbn__SchemaVersion != null;
+        public void ResetSchemaVersion() => __pbn__SchemaVersion = null;
+        private byte[] __pbn__SchemaVersion;
+
+        [global::ProtoBuf.ProtoMember(17, Name = @"partition_key_b64_encoded")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool PartitionKeyB64Encoded
+        {
+            get { return __pbn__PartitionKeyB64Encoded ?? false; }
+            set { __pbn__PartitionKeyB64Encoded = value; }
+        }
+        public bool ShouldSerializePartitionKeyB64Encoded() => __pbn__PartitionKeyB64Encoded != null;
+        public void ResetPartitionKeyB64Encoded() => __pbn__PartitionKeyB64Encoded = null;
+        private bool? __pbn__PartitionKeyB64Encoded;
+
+        [global::ProtoBuf.ProtoMember(18, Name = @"ordering_key")]
+        public byte[] OrderingKey
+        {
+            get { return __pbn__OrderingKey; }
+            set { __pbn__OrderingKey = value; }
+        }
+        public bool ShouldSerializeOrderingKey() => __pbn__OrderingKey != null;
+        public void ResetOrderingKey() => __pbn__OrderingKey = null;
+        private byte[] __pbn__OrderingKey;
+
+        [global::ProtoBuf.ProtoMember(19, Name = @"deliver_at_time")]
+        public long DeliverAtTime
+        {
+            get { return __pbn__DeliverAtTime.GetValueOrDefault(); }
+            set { __pbn__DeliverAtTime = value; }
+        }
+        public bool ShouldSerializeDeliverAtTime() => __pbn__DeliverAtTime != null;
+        public void ResetDeliverAtTime() => __pbn__DeliverAtTime = null;
+        private long? __pbn__DeliverAtTime;
+
+        [global::ProtoBuf.ProtoMember(20, Name = @"marker_type")]
+        public int MarkerType
+        {
+            get { return __pbn__MarkerType.GetValueOrDefault(); }
+            set { __pbn__MarkerType = value; }
+        }
+        public bool ShouldSerializeMarkerType() => __pbn__MarkerType != null;
+        public void ResetMarkerType() => __pbn__MarkerType = null;
+        private int? __pbn__MarkerType;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class SingleMessageMetadata : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"properties")]
+        public global::System.Collections.Generic.List<KeyValue> Properties { get; } = new global::System.Collections.Generic.List<KeyValue>();
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"partition_key")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string PartitionKey
+        {
+            get { return __pbn__PartitionKey ?? ""; }
+            set { __pbn__PartitionKey = value; }
+        }
+        public bool ShouldSerializePartitionKey() => __pbn__PartitionKey != null;
+        public void ResetPartitionKey() => __pbn__PartitionKey = null;
+        private string __pbn__PartitionKey;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"payload_size", IsRequired = true)]
+        public int PayloadSize { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"compacted_out")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool CompactedOut
+        {
+            get { return __pbn__CompactedOut ?? false; }
+            set { __pbn__CompactedOut = value; }
+        }
+        public bool ShouldSerializeCompactedOut() => __pbn__CompactedOut != null;
+        public void ResetCompactedOut() => __pbn__CompactedOut = null;
+        private bool? __pbn__CompactedOut;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"event_time")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public ulong EventTime
+        {
+            get { return __pbn__EventTime ?? 0; }
+            set { __pbn__EventTime = value; }
+        }
+        public bool ShouldSerializeEventTime() => __pbn__EventTime != null;
+        public void ResetEventTime() => __pbn__EventTime = null;
+        private ulong? __pbn__EventTime;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"partition_key_b64_encoded")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool PartitionKeyB64Encoded
+        {
+            get { return __pbn__PartitionKeyB64Encoded ?? false; }
+            set { __pbn__PartitionKeyB64Encoded = value; }
+        }
+        public bool ShouldSerializePartitionKeyB64Encoded() => __pbn__PartitionKeyB64Encoded != null;
+        public void ResetPartitionKeyB64Encoded() => __pbn__PartitionKeyB64Encoded = null;
+        private bool? __pbn__PartitionKeyB64Encoded;
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"ordering_key")]
+        public byte[] OrderingKey
+        {
+            get { return __pbn__OrderingKey; }
+            set { __pbn__OrderingKey = value; }
+        }
+        public bool ShouldSerializeOrderingKey() => __pbn__OrderingKey != null;
+        public void ResetOrderingKey() => __pbn__OrderingKey = null;
+        private byte[] __pbn__OrderingKey;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandConnect : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"client_version", IsRequired = true)]
+        public string ClientVersion { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"auth_method")]
+        [global::System.ComponentModel.DefaultValue(AuthMethod.AuthMethodNone)]
+        public AuthMethod AuthMethod
+        {
+            get { return __pbn__AuthMethod ?? AuthMethod.AuthMethodNone; }
+            set { __pbn__AuthMethod = value; }
+        }
+        public bool ShouldSerializeAuthMethod() => __pbn__AuthMethod != null;
+        public void ResetAuthMethod() => __pbn__AuthMethod = null;
+        private AuthMethod? __pbn__AuthMethod;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"auth_method_name")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string AuthMethodName
+        {
+            get { return __pbn__AuthMethodName ?? ""; }
+            set { __pbn__AuthMethodName = value; }
+        }
+        public bool ShouldSerializeAuthMethodName() => __pbn__AuthMethodName != null;
+        public void ResetAuthMethodName() => __pbn__AuthMethodName = null;
+        private string __pbn__AuthMethodName;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"auth_data")]
+        public byte[] AuthData
+        {
+            get { return __pbn__AuthData; }
+            set { __pbn__AuthData = value; }
+        }
+        public bool ShouldSerializeAuthData() => __pbn__AuthData != null;
+        public void ResetAuthData() => __pbn__AuthData = null;
+        private byte[] __pbn__AuthData;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"protocol_version")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public int ProtocolVersion
+        {
+            get { return __pbn__ProtocolVersion ?? 0; }
+            set { __pbn__ProtocolVersion = value; }
+        }
+        public bool ShouldSerializeProtocolVersion() => __pbn__ProtocolVersion != null;
+        public void ResetProtocolVersion() => __pbn__ProtocolVersion = null;
+        private int? __pbn__ProtocolVersion;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"proxy_to_broker_url")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ProxyToBrokerUrl
+        {
+            get { return __pbn__ProxyToBrokerUrl ?? ""; }
+            set { __pbn__ProxyToBrokerUrl = value; }
+        }
+        public bool ShouldSerializeProxyToBrokerUrl() => __pbn__ProxyToBrokerUrl != null;
+        public void ResetProxyToBrokerUrl() => __pbn__ProxyToBrokerUrl = null;
+        private string __pbn__ProxyToBrokerUrl;
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"original_principal")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalPrincipal
+        {
+            get { return __pbn__OriginalPrincipal ?? ""; }
+            set { __pbn__OriginalPrincipal = value; }
+        }
+        public bool ShouldSerializeOriginalPrincipal() => __pbn__OriginalPrincipal != null;
+        public void ResetOriginalPrincipal() => __pbn__OriginalPrincipal = null;
+        private string __pbn__OriginalPrincipal;
+
+        [global::ProtoBuf.ProtoMember(8, Name = @"original_auth_data")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthData
+        {
+            get { return __pbn__OriginalAuthData ?? ""; }
+            set { __pbn__OriginalAuthData = value; }
+        }
+        public bool ShouldSerializeOriginalAuthData() => __pbn__OriginalAuthData != null;
+        public void ResetOriginalAuthData() => __pbn__OriginalAuthData = null;
+        private string __pbn__OriginalAuthData;
+
+        [global::ProtoBuf.ProtoMember(9, Name = @"original_auth_method")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthMethod
+        {
+            get { return __pbn__OriginalAuthMethod ?? ""; }
+            set { __pbn__OriginalAuthMethod = value; }
+        }
+        public bool ShouldSerializeOriginalAuthMethod() => __pbn__OriginalAuthMethod != null;
+        public void ResetOriginalAuthMethod() => __pbn__OriginalAuthMethod = null;
+        private string __pbn__OriginalAuthMethod;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandConnected : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"server_version", IsRequired = true)]
+        public string ServerVersion { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"protocol_version")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public int ProtocolVersion
+        {
+            get { return __pbn__ProtocolVersion ?? 0; }
+            set { __pbn__ProtocolVersion = value; }
+        }
+        public bool ShouldSerializeProtocolVersion() => __pbn__ProtocolVersion != null;
+        public void ResetProtocolVersion() => __pbn__ProtocolVersion = null;
+        private int? __pbn__ProtocolVersion;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"max_message_size")]
+        public int MaxMessageSize
+        {
+            get { return __pbn__MaxMessageSize.GetValueOrDefault(); }
+            set { __pbn__MaxMessageSize = value; }
+        }
+        public bool ShouldSerializeMaxMessageSize() => __pbn__MaxMessageSize != null;
+        public void ResetMaxMessageSize() => __pbn__MaxMessageSize = null;
+        private int? __pbn__MaxMessageSize;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandAuthResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"client_version")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ClientVersion
+        {
+            get { return __pbn__ClientVersion ?? ""; }
+            set { __pbn__ClientVersion = value; }
+        }
+        public bool ShouldSerializeClientVersion() => __pbn__ClientVersion != null;
+        public void ResetClientVersion() => __pbn__ClientVersion = null;
+        private string __pbn__ClientVersion;
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"response")]
+        public AuthData Response { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"protocol_version")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public int ProtocolVersion
+        {
+            get { return __pbn__ProtocolVersion ?? 0; }
+            set { __pbn__ProtocolVersion = value; }
+        }
+        public bool ShouldSerializeProtocolVersion() => __pbn__ProtocolVersion != null;
+        public void ResetProtocolVersion() => __pbn__ProtocolVersion = null;
+        private int? __pbn__ProtocolVersion;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandAuthChallenge : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"server_version")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ServerVersion
+        {
+            get { return __pbn__ServerVersion ?? ""; }
+            set { __pbn__ServerVersion = value; }
+        }
+        public bool ShouldSerializeServerVersion() => __pbn__ServerVersion != null;
+        public void ResetServerVersion() => __pbn__ServerVersion = null;
+        private string __pbn__ServerVersion;
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"challenge")]
+        public AuthData Challenge { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"protocol_version")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public int ProtocolVersion
+        {
+            get { return __pbn__ProtocolVersion ?? 0; }
+            set { __pbn__ProtocolVersion = value; }
+        }
+        public bool ShouldSerializeProtocolVersion() => __pbn__ProtocolVersion != null;
+        public void ResetProtocolVersion() => __pbn__ProtocolVersion = null;
+        private int? __pbn__ProtocolVersion;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class AuthData : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"auth_method_name")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string AuthMethodName
+        {
+            get { return __pbn__AuthMethodName ?? ""; }
+            set { __pbn__AuthMethodName = value; }
+        }
+        public bool ShouldSerializeAuthMethodName() => __pbn__AuthMethodName != null;
+        public void ResetAuthMethodName() => __pbn__AuthMethodName = null;
+        private string __pbn__AuthMethodName;
+
+        [global::ProtoBuf.ProtoMember(2)]
+        public byte[] auth_data
+        {
+            get { return __pbn__auth_data; }
+            set { __pbn__auth_data = value; }
+        }
+        public bool ShouldSerializeauth_data() => __pbn__auth_data != null;
+        public void Resetauth_data() => __pbn__auth_data = null;
+        private byte[] __pbn__auth_data;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSubscribe : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"topic", IsRequired = true)]
+        public string Topic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"subscription", IsRequired = true)]
+        public string Subscription { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, IsRequired = true)]
+        public SubType Type { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"consumer_name")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ConsumerName
+        {
+            get { return __pbn__ConsumerName ?? ""; }
+            set { __pbn__ConsumerName = value; }
+        }
+        public bool ShouldSerializeConsumerName() => __pbn__ConsumerName != null;
+        public void ResetConsumerName() => __pbn__ConsumerName = null;
+        private string __pbn__ConsumerName;
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"priority_level")]
+        public int PriorityLevel
+        {
+            get { return __pbn__PriorityLevel.GetValueOrDefault(); }
+            set { __pbn__PriorityLevel = value; }
+        }
+        public bool ShouldSerializePriorityLevel() => __pbn__PriorityLevel != null;
+        public void ResetPriorityLevel() => __pbn__PriorityLevel = null;
+        private int? __pbn__PriorityLevel;
+
+        [global::ProtoBuf.ProtoMember(8, Name = @"durable")]
+        [global::System.ComponentModel.DefaultValue(true)]
+        public bool Durable
+        {
+            get { return __pbn__Durable ?? true; }
+            set { __pbn__Durable = value; }
+        }
+        public bool ShouldSerializeDurable() => __pbn__Durable != null;
+        public void ResetDurable() => __pbn__Durable = null;
+        private bool? __pbn__Durable;
+
+        [global::ProtoBuf.ProtoMember(9, Name = @"start_message_id")]
+        public MessageIdData StartMessageId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(10, Name = @"metadata")]
+        public global::System.Collections.Generic.List<KeyValue> Metadatas { get; } = new global::System.Collections.Generic.List<KeyValue>();
+
+        [global::ProtoBuf.ProtoMember(11, Name = @"read_compacted")]
+        public bool ReadCompacted
+        {
+            get { return __pbn__ReadCompacted.GetValueOrDefault(); }
+            set { __pbn__ReadCompacted = value; }
+        }
+        public bool ShouldSerializeReadCompacted() => __pbn__ReadCompacted != null;
+        public void ResetReadCompacted() => __pbn__ReadCompacted = null;
+        private bool? __pbn__ReadCompacted;
+
+        [global::ProtoBuf.ProtoMember(12, Name = @"schema")]
+        public Schema Schema { get; set; }
+
+        [global::ProtoBuf.ProtoMember(13)]
+        [global::System.ComponentModel.DefaultValue(InitialPosition.Latest)]
+        public InitialPosition initialPosition
+        {
+            get { return __pbn__initialPosition ?? InitialPosition.Latest; }
+            set { __pbn__initialPosition = value; }
+        }
+        public bool ShouldSerializeinitialPosition() => __pbn__initialPosition != null;
+        public void ResetinitialPosition() => __pbn__initialPosition = null;
+        private InitialPosition? __pbn__initialPosition;
+
+        [global::ProtoBuf.ProtoMember(14, Name = @"replicate_subscription_state")]
+        public bool ReplicateSubscriptionState
+        {
+            get { return __pbn__ReplicateSubscriptionState.GetValueOrDefault(); }
+            set { __pbn__ReplicateSubscriptionState = value; }
+        }
+        public bool ShouldSerializeReplicateSubscriptionState() => __pbn__ReplicateSubscriptionState != null;
+        public void ResetReplicateSubscriptionState() => __pbn__ReplicateSubscriptionState = null;
+        private bool? __pbn__ReplicateSubscriptionState;
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum SubType
+        {
+            Exclusive = 0,
+            Shared = 1,
+            Failover = 2,
+            [global::ProtoBuf.ProtoEnum(Name = @"Key_Shared")]
+            KeyShared = 3,
+        }
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum InitialPosition
+        {
+            Latest = 0,
+            Earliest = 1,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandPartitionedTopicMetadata : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"topic", IsRequired = true)]
+        public string Topic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"original_principal")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalPrincipal
+        {
+            get { return __pbn__OriginalPrincipal ?? ""; }
+            set { __pbn__OriginalPrincipal = value; }
+        }
+        public bool ShouldSerializeOriginalPrincipal() => __pbn__OriginalPrincipal != null;
+        public void ResetOriginalPrincipal() => __pbn__OriginalPrincipal = null;
+        private string __pbn__OriginalPrincipal;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"original_auth_data")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthData
+        {
+            get { return __pbn__OriginalAuthData ?? ""; }
+            set { __pbn__OriginalAuthData = value; }
+        }
+        public bool ShouldSerializeOriginalAuthData() => __pbn__OriginalAuthData != null;
+        public void ResetOriginalAuthData() => __pbn__OriginalAuthData = null;
+        private string __pbn__OriginalAuthData;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"original_auth_method")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthMethod
+        {
+            get { return __pbn__OriginalAuthMethod ?? ""; }
+            set { __pbn__OriginalAuthMethod = value; }
+        }
+        public bool ShouldSerializeOriginalAuthMethod() => __pbn__OriginalAuthMethod != null;
+        public void ResetOriginalAuthMethod() => __pbn__OriginalAuthMethod = null;
+        private string __pbn__OriginalAuthMethod;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandPartitionedTopicMetadataResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"partitions")]
+        public uint Partitions
+        {
+            get { return __pbn__Partitions.GetValueOrDefault(); }
+            set { __pbn__Partitions = value; }
+        }
+        public bool ShouldSerializePartitions() => __pbn__Partitions != null;
+        public void ResetPartitions() => __pbn__Partitions = null;
+        private uint? __pbn__Partitions;
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"response")]
+        [global::System.ComponentModel.DefaultValue(LookupType.Success)]
+        public LookupType Response
+        {
+            get { return __pbn__Response ?? LookupType.Success; }
+            set { __pbn__Response = value; }
+        }
+        public bool ShouldSerializeResponse() => __pbn__Response != null;
+        public void ResetResponse() => __pbn__Response = null;
+        private LookupType? __pbn__Response;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"error")]
+        [global::System.ComponentModel.DefaultValue(ServerError.UnknownError)]
+        public ServerError Error
+        {
+            get { return __pbn__Error ?? ServerError.UnknownError; }
+            set { __pbn__Error = value; }
+        }
+        public bool ShouldSerializeError() => __pbn__Error != null;
+        public void ResetError() => __pbn__Error = null;
+        private ServerError? __pbn__Error;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"message")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string Message
+        {
+            get { return __pbn__Message ?? ""; }
+            set { __pbn__Message = value; }
+        }
+        public bool ShouldSerializeMessage() => __pbn__Message != null;
+        public void ResetMessage() => __pbn__Message = null;
+        private string __pbn__Message;
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum LookupType
+        {
+            Success = 0,
+            Failed = 1,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandLookupTopic : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"topic", IsRequired = true)]
+        public string Topic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"authoritative")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool Authoritative
+        {
+            get { return __pbn__Authoritative ?? false; }
+            set { __pbn__Authoritative = value; }
+        }
+        public bool ShouldSerializeAuthoritative() => __pbn__Authoritative != null;
+        public void ResetAuthoritative() => __pbn__Authoritative = null;
+        private bool? __pbn__Authoritative;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"original_principal")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalPrincipal
+        {
+            get { return __pbn__OriginalPrincipal ?? ""; }
+            set { __pbn__OriginalPrincipal = value; }
+        }
+        public bool ShouldSerializeOriginalPrincipal() => __pbn__OriginalPrincipal != null;
+        public void ResetOriginalPrincipal() => __pbn__OriginalPrincipal = null;
+        private string __pbn__OriginalPrincipal;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"original_auth_data")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthData
+        {
+            get { return __pbn__OriginalAuthData ?? ""; }
+            set { __pbn__OriginalAuthData = value; }
+        }
+        public bool ShouldSerializeOriginalAuthData() => __pbn__OriginalAuthData != null;
+        public void ResetOriginalAuthData() => __pbn__OriginalAuthData = null;
+        private string __pbn__OriginalAuthData;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"original_auth_method")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string OriginalAuthMethod
+        {
+            get { return __pbn__OriginalAuthMethod ?? ""; }
+            set { __pbn__OriginalAuthMethod = value; }
+        }
+        public bool ShouldSerializeOriginalAuthMethod() => __pbn__OriginalAuthMethod != null;
+        public void ResetOriginalAuthMethod() => __pbn__OriginalAuthMethod = null;
+        private string __pbn__OriginalAuthMethod;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandLookupTopicResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1)]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string BrokerServiceUrl
+        {
+            get { return __pbn__brokerServiceUrl ?? ""; }
+            set { __pbn__brokerServiceUrl = value; }
+        }
+        public bool ShouldSerializebrokerServiceUrl() => __pbn__brokerServiceUrl != null;
+        public void ResetbrokerServiceUrl() => __pbn__brokerServiceUrl = null;
+        private string __pbn__brokerServiceUrl;
+
+        [global::ProtoBuf.ProtoMember(2)]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string BrokerServiceUrlTls
+        {
+            get { return __pbn__brokerServiceUrlTls ?? ""; }
+            set { __pbn__brokerServiceUrlTls = value; }
+        }
+        public bool ShouldSerializebrokerServiceUrlTls() => __pbn__brokerServiceUrlTls != null;
+        public void ResetbrokerServiceUrlTls() => __pbn__brokerServiceUrlTls = null;
+        private string __pbn__brokerServiceUrlTls;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"response")]
+        [global::System.ComponentModel.DefaultValue(LookupType.Redirect)]
+        public LookupType Response
+        {
+            get { return __pbn__Response ?? LookupType.Redirect; }
+            set { __pbn__Response = value; }
+        }
+        public bool ShouldSerializeResponse() => __pbn__Response != null;
+        public void ResetResponse() => __pbn__Response = null;
+        private LookupType? __pbn__Response;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"authoritative")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool Authoritative
+        {
+            get { return __pbn__Authoritative ?? false; }
+            set { __pbn__Authoritative = value; }
+        }
+        public bool ShouldSerializeAuthoritative() => __pbn__Authoritative != null;
+        public void ResetAuthoritative() => __pbn__Authoritative = null;
+        private bool? __pbn__Authoritative;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"error")]
+        [global::System.ComponentModel.DefaultValue(ServerError.UnknownError)]
+        public ServerError Error
+        {
+            get { return __pbn__Error ?? ServerError.UnknownError; }
+            set { __pbn__Error = value; }
+        }
+        public bool ShouldSerializeError() => __pbn__Error != null;
+        public void ResetError() => __pbn__Error = null;
+        private ServerError? __pbn__Error;
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"message")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string Message
+        {
+            get { return __pbn__Message ?? ""; }
+            set { __pbn__Message = value; }
+        }
+        public bool ShouldSerializeMessage() => __pbn__Message != null;
+        public void ResetMessage() => __pbn__Message = null;
+        private string __pbn__Message;
+
+        [global::ProtoBuf.ProtoMember(8, Name = @"proxy_through_service_url")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool ProxyThroughServiceUrl
+        {
+            get { return __pbn__ProxyThroughServiceUrl ?? false; }
+            set { __pbn__ProxyThroughServiceUrl = value; }
+        }
+        public bool ShouldSerializeProxyThroughServiceUrl() => __pbn__ProxyThroughServiceUrl != null;
+        public void ResetProxyThroughServiceUrl() => __pbn__ProxyThroughServiceUrl = null;
+        private bool? __pbn__ProxyThroughServiceUrl;
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum LookupType
+        {
+            Redirect = 0,
+            Connect = 1,
+            Failed = 2,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandProducer : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"topic", IsRequired = true)]
+        public string Topic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"producer_id", IsRequired = true)]
+        public ulong ProducerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"producer_name")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ProducerName
+        {
+            get { return __pbn__ProducerName ?? ""; }
+            set { __pbn__ProducerName = value; }
+        }
+        public bool ShouldSerializeProducerName() => __pbn__ProducerName != null;
+        public void ResetProducerName() => __pbn__ProducerName = null;
+        private string __pbn__ProducerName;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"encrypted")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool Encrypted
+        {
+            get { return __pbn__Encrypted ?? false; }
+            set { __pbn__Encrypted = value; }
+        }
+        public bool ShouldSerializeEncrypted() => __pbn__Encrypted != null;
+        public void ResetEncrypted() => __pbn__Encrypted = null;
+        private bool? __pbn__Encrypted;
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"metadata")]
+        public global::System.Collections.Generic.List<KeyValue> Metadatas { get; } = new global::System.Collections.Generic.List<KeyValue>();
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"schema")]
+        public Schema Schema { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSend : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"producer_id", IsRequired = true)]
+        public ulong ProducerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"sequence_id", IsRequired = true)]
+        public ulong SequenceId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"num_messages")]
+        [global::System.ComponentModel.DefaultValue(1)]
+        public int NumMessages
+        {
+            get { return __pbn__NumMessages ?? 1; }
+            set { __pbn__NumMessages = value; }
+        }
+        public bool ShouldSerializeNumMessages() => __pbn__NumMessages != null;
+        public void ResetNumMessages() => __pbn__NumMessages = null;
+        private int? __pbn__NumMessages;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSendReceipt : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"producer_id", IsRequired = true)]
+        public ulong ProducerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"sequence_id", IsRequired = true)]
+        public ulong SequenceId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"message_id")]
+        public MessageIdData MessageId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSendError : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"producer_id", IsRequired = true)]
+        public ulong ProducerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"sequence_id", IsRequired = true)]
+        public ulong SequenceId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"error", IsRequired = true)]
+        public ServerError Error { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"message", IsRequired = true)]
+        public string Message { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandMessage : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"message_id", IsRequired = true)]
+        public MessageIdData MessageId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"redelivery_count")]
+        [global::System.ComponentModel.DefaultValue(0)]
+        public uint RedeliveryCount
+        {
+            get { return __pbn__RedeliveryCount ?? 0; }
+            set { __pbn__RedeliveryCount = value; }
+        }
+        public bool ShouldSerializeRedeliveryCount() => __pbn__RedeliveryCount != null;
+        public void ResetRedeliveryCount() => __pbn__RedeliveryCount = null;
+        private uint? __pbn__RedeliveryCount;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandAck : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, IsRequired = true)]
+        public AckType Type { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"message_id")]
+        public global::System.Collections.Generic.List<MessageIdData> MessageIds { get; } = new global::System.Collections.Generic.List<MessageIdData>(1);
+
+        [global::ProtoBuf.ProtoMember(4)]
+        [global::System.ComponentModel.DefaultValue(ValidationError.UncompressedSizeCorruption)]
+        public ValidationError validation_error
+        {
+            get { return __pbn__validation_error ?? ValidationError.UncompressedSizeCorruption; }
+            set { __pbn__validation_error = value; }
+        }
+        public bool ShouldSerializevalidation_error() => __pbn__validation_error != null;
+        public void Resetvalidation_error() => __pbn__validation_error = null;
+        private ValidationError? __pbn__validation_error;
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"properties")]
+        public global::System.Collections.Generic.List<KeyLongValue> Properties { get; } = new global::System.Collections.Generic.List<KeyLongValue>();
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum AckType
+        {
+            Individual = 0,
+            Cumulative = 1,
+        }
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum ValidationError
+        {
+            UncompressedSizeCorruption = 0,
+            DecompressionError = 1,
+            ChecksumMismatch = 2,
+            BatchDeSerializeError = 3,
+            DecryptionError = 4,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandActiveConsumerChange : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"is_active")]
+        [global::System.ComponentModel.DefaultValue(false)]
+        public bool IsActive
+        {
+            get { return __pbn__IsActive ?? false; }
+            set { __pbn__IsActive = value; }
+        }
+        public bool ShouldSerializeIsActive() => __pbn__IsActive != null;
+        public void ResetIsActive() => __pbn__IsActive = null;
+        private bool? __pbn__IsActive;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandFlow : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, IsRequired = true)]
+        public uint MessagePermits { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandUnsubscribe : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSeek : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"message_id")]
+        public MessageIdData MessageId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"message_publish_time")]
+        public ulong MessagePublishTime
+        {
+            get { return __pbn__MessagePublishTime.GetValueOrDefault(); }
+            set { __pbn__MessagePublishTime = value; }
+        }
+        public bool ShouldSerializeMessagePublishTime() => __pbn__MessagePublishTime != null;
+        public void ResetMessagePublishTime() => __pbn__MessagePublishTime = null;
+        private ulong? __pbn__MessagePublishTime;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandReachedEndOfTopic : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandCloseProducer : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"producer_id", IsRequired = true)]
+        public ulong ProducerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandCloseConsumer : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandRedeliverUnacknowledgedMessages : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"message_ids")]
+        public global::System.Collections.Generic.List<MessageIdData> MessageIds { get; } = new global::System.Collections.Generic.List<MessageIdData>();
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandSuccess : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"schema")]
+        public Schema Schema { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandProducerSuccess : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"producer_name", IsRequired = true)]
+        public string ProducerName { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"last_sequence_id")]
+        [global::System.ComponentModel.DefaultValue(-1)]
+        public long LastSequenceId
+        {
+            get { return __pbn__LastSequenceId ?? -1; }
+            set { __pbn__LastSequenceId = value; }
+        }
+        public bool ShouldSerializeLastSequenceId() => __pbn__LastSequenceId != null;
+        public void ResetLastSequenceId() => __pbn__LastSequenceId = null;
+        private long? __pbn__LastSequenceId;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"schema_version")]
+        public byte[] SchemaVersion
+        {
+            get { return __pbn__SchemaVersion; }
+            set { __pbn__SchemaVersion = value; }
+        }
+        public bool ShouldSerializeSchemaVersion() => __pbn__SchemaVersion != null;
+        public void ResetSchemaVersion() => __pbn__SchemaVersion = null;
+        private byte[] __pbn__SchemaVersion;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandError : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"error", IsRequired = true)]
+        public ServerError Error { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"message", IsRequired = true)]
+        public string Message { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandPing : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandPong : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandConsumerStats : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandConsumerStatsResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"error_code")]
+        [global::System.ComponentModel.DefaultValue(ServerError.UnknownError)]
+        public ServerError ErrorCode
+        {
+            get { return __pbn__ErrorCode ?? ServerError.UnknownError; }
+            set { __pbn__ErrorCode = value; }
+        }
+        public bool ShouldSerializeErrorCode() => __pbn__ErrorCode != null;
+        public void ResetErrorCode() => __pbn__ErrorCode = null;
+        private ServerError? __pbn__ErrorCode;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"error_message")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ErrorMessage
+        {
+            get { return __pbn__ErrorMessage ?? ""; }
+            set { __pbn__ErrorMessage = value; }
+        }
+        public bool ShouldSerializeErrorMessage() => __pbn__ErrorMessage != null;
+        public void ResetErrorMessage() => __pbn__ErrorMessage = null;
+        private string __pbn__ErrorMessage;
+
+        [global::ProtoBuf.ProtoMember(4)]
+        public double MsgRateOut
+        {
+            get { return __pbn__msgRateOut.GetValueOrDefault(); }
+            set { __pbn__msgRateOut = value; }
+        }
+        public bool ShouldSerializemsgRateOut() => __pbn__msgRateOut != null;
+        public void ResetmsgRateOut() => __pbn__msgRateOut = null;
+        private double? __pbn__msgRateOut;
+
+        [global::ProtoBuf.ProtoMember(5)]
+        public double MsgThroughputOut
+        {
+            get { return __pbn__msgThroughputOut.GetValueOrDefault(); }
+            set { __pbn__msgThroughputOut = value; }
+        }
+        public bool ShouldSerializemsgThroughputOut() => __pbn__msgThroughputOut != null;
+        public void ResetmsgThroughputOut() => __pbn__msgThroughputOut = null;
+        private double? __pbn__msgThroughputOut;
+
+        [global::ProtoBuf.ProtoMember(6)]
+        public double MsgRateRedeliver
+        {
+            get { return __pbn__msgRateRedeliver.GetValueOrDefault(); }
+            set { __pbn__msgRateRedeliver = value; }
+        }
+        public bool ShouldSerializemsgRateRedeliver() => __pbn__msgRateRedeliver != null;
+        public void ResetmsgRateRedeliver() => __pbn__msgRateRedeliver = null;
+        private double? __pbn__msgRateRedeliver;
+
+        [global::ProtoBuf.ProtoMember(7)]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ConsumerName
+        {
+            get { return __pbn__consumerName ?? ""; }
+            set { __pbn__consumerName = value; }
+        }
+        public bool ShouldSerializeconsumerName() => __pbn__consumerName != null;
+        public void ResetconsumerName() => __pbn__consumerName = null;
+        private string __pbn__consumerName;
+
+        [global::ProtoBuf.ProtoMember(8)]
+        public ulong AvailablePermits
+        {
+            get { return __pbn__availablePermits.GetValueOrDefault(); }
+            set { __pbn__availablePermits = value; }
+        }
+        public bool ShouldSerializeavailablePermits() => __pbn__availablePermits != null;
+        public void ResetavailablePermits() => __pbn__availablePermits = null;
+        private ulong? __pbn__availablePermits;
+
+        [global::ProtoBuf.ProtoMember(9)]
+        public ulong UnackedMessages
+        {
+            get { return __pbn__unackedMessages.GetValueOrDefault(); }
+            set { __pbn__unackedMessages = value; }
+        }
+        public bool ShouldSerializeunackedMessages() => __pbn__unackedMessages != null;
+        public void ResetunackedMessages() => __pbn__unackedMessages = null;
+        private ulong? __pbn__unackedMessages;
+
+        [global::ProtoBuf.ProtoMember(10)]
+        public bool BlockedConsumerOnUnackedMsgs
+        {
+            get { return __pbn__blockedConsumerOnUnackedMsgs.GetValueOrDefault(); }
+            set { __pbn__blockedConsumerOnUnackedMsgs = value; }
+        }
+        public bool ShouldSerializeblockedConsumerOnUnackedMsgs() => __pbn__blockedConsumerOnUnackedMsgs != null;
+        public void ResetblockedConsumerOnUnackedMsgs() => __pbn__blockedConsumerOnUnackedMsgs = null;
+        private bool? __pbn__blockedConsumerOnUnackedMsgs;
+
+        [global::ProtoBuf.ProtoMember(11, Name = @"address")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string Address
+        {
+            get { return __pbn__Address ?? ""; }
+            set { __pbn__Address = value; }
+        }
+        public bool ShouldSerializeAddress() => __pbn__Address != null;
+        public void ResetAddress() => __pbn__Address = null;
+        private string __pbn__Address;
+
+        [global::ProtoBuf.ProtoMember(12)]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ConnectedSince
+        {
+            get { return __pbn__connectedSince ?? ""; }
+            set { __pbn__connectedSince = value; }
+        }
+        public bool ShouldSerializeconnectedSince() => __pbn__connectedSince != null;
+        public void ResetconnectedSince() => __pbn__connectedSince = null;
+        private string __pbn__connectedSince;
+
+        [global::ProtoBuf.ProtoMember(13, Name = @"type")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string Type
+        {
+            get { return __pbn__Type ?? ""; }
+            set { __pbn__Type = value; }
+        }
+        public bool ShouldSerializeType() => __pbn__Type != null;
+        public void ResetType() => __pbn__Type = null;
+        private string __pbn__Type;
+
+        [global::ProtoBuf.ProtoMember(14)]
+        public double MsgRateExpired
+        {
+            get { return __pbn__msgRateExpired.GetValueOrDefault(); }
+            set { __pbn__msgRateExpired = value; }
+        }
+        public bool ShouldSerializemsgRateExpired() => __pbn__msgRateExpired != null;
+        public void ResetmsgRateExpired() => __pbn__msgRateExpired = null;
+        private double? __pbn__msgRateExpired;
+
+        [global::ProtoBuf.ProtoMember(15)]
+        public ulong MsgBacklog
+        {
+            get { return __pbn__msgBacklog.GetValueOrDefault(); }
+            set { __pbn__msgBacklog = value; }
+        }
+        public bool ShouldSerializemsgBacklog() => __pbn__msgBacklog != null;
+        public void ResetmsgBacklog() => __pbn__msgBacklog = null;
+        private ulong? __pbn__msgBacklog;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetLastMessageId : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"consumer_id", IsRequired = true)]
+        public ulong ConsumerId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetLastMessageIdResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"last_message_id", IsRequired = true)]
+        public MessageIdData LastMessageId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetTopicsOfNamespace : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"namespace", IsRequired = true)]
+        public string Namespace { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3)]
+        [global::System.ComponentModel.DefaultValue(Mode.Persistent)]
+        public Mode mode
+        {
+            get { return __pbn__mode ?? Mode.Persistent; }
+            set { __pbn__mode = value; }
+        }
+        public bool ShouldSerializemode() => __pbn__mode != null;
+        public void Resetmode() => __pbn__mode = null;
+        private Mode? __pbn__mode;
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum Mode
+        {
+            [global::ProtoBuf.ProtoEnum(Name = @"PERSISTENT")]
+            Persistent = 0,
+            [global::ProtoBuf.ProtoEnum(Name = @"NON_PERSISTENT")]
+            NonPersistent = 1,
+            [global::ProtoBuf.ProtoEnum(Name = @"ALL")]
+            All = 2,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetTopicsOfNamespaceResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"topics")]
+        public global::System.Collections.Generic.List<string> Topics { get; } = new global::System.Collections.Generic.List<string>();
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetSchema : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"topic", IsRequired = true)]
+        public string Topic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"schema_version")]
+        public byte[] SchemaVersion
+        {
+            get { return __pbn__SchemaVersion; }
+            set { __pbn__SchemaVersion = value; }
+        }
+        public bool ShouldSerializeSchemaVersion() => __pbn__SchemaVersion != null;
+        public void ResetSchemaVersion() => __pbn__SchemaVersion = null;
+        private byte[] __pbn__SchemaVersion;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class CommandGetSchemaResponse : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, Name = @"request_id", IsRequired = true)]
+        public ulong RequestId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"error_code")]
+        [global::System.ComponentModel.DefaultValue(ServerError.UnknownError)]
+        public ServerError ErrorCode
+        {
+            get { return __pbn__ErrorCode ?? ServerError.UnknownError; }
+            set { __pbn__ErrorCode = value; }
+        }
+        public bool ShouldSerializeErrorCode() => __pbn__ErrorCode != null;
+        public void ResetErrorCode() => __pbn__ErrorCode = null;
+        private ServerError? __pbn__ErrorCode;
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"error_message")]
+        [global::System.ComponentModel.DefaultValue("")]
+        public string ErrorMessage
+        {
+            get { return __pbn__ErrorMessage ?? ""; }
+            set { __pbn__ErrorMessage = value; }
+        }
+        public bool ShouldSerializeErrorMessage() => __pbn__ErrorMessage != null;
+        public void ResetErrorMessage() => __pbn__ErrorMessage = null;
+        private string __pbn__ErrorMessage;
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"schema")]
+        public Schema Schema { get; set; }
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"schema_version")]
+        public byte[] SchemaVersion
+        {
+            get { return __pbn__SchemaVersion; }
+            set { __pbn__SchemaVersion = value; }
+        }
+        public bool ShouldSerializeSchemaVersion() => __pbn__SchemaVersion != null;
+        public void ResetSchemaVersion() => __pbn__SchemaVersion = null;
+        private byte[] __pbn__SchemaVersion;
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public partial class BaseCommand : global::ProtoBuf.IExtensible
+    {
+        private global::ProtoBuf.IExtension __pbn__extensionData;
+        global::ProtoBuf.IExtension global::ProtoBuf.IExtensible.GetExtensionObject(bool createIfMissing)
+            => global::ProtoBuf.Extensible.GetExtensionObject(ref __pbn__extensionData, createIfMissing);
+
+        [global::ProtoBuf.ProtoMember(1, IsRequired = true)]
+        public Type CommandType { get; set; } = Type.Connect;
+
+        [global::ProtoBuf.ProtoMember(2, Name = @"connect")]
+        public CommandConnect Connect { get; set; }
+
+        [global::ProtoBuf.ProtoMember(3, Name = @"connected")]
+        public CommandConnected Connected { get; set; }
+
+        [global::ProtoBuf.ProtoMember(4, Name = @"subscribe")]
+        public CommandSubscribe Subscribe { get; set; }
+
+        [global::ProtoBuf.ProtoMember(5, Name = @"producer")]
+        public CommandProducer Producer { get; set; }
+
+        [global::ProtoBuf.ProtoMember(6, Name = @"send")]
+        public CommandSend Send { get; set; }
+
+        [global::ProtoBuf.ProtoMember(7, Name = @"send_receipt")]
+        public CommandSendReceipt SendReceipt { get; set; }
+
+        [global::ProtoBuf.ProtoMember(8, Name = @"send_error")]
+        public CommandSendError SendError { get; set; }
+
+        [global::ProtoBuf.ProtoMember(9, Name = @"message")]
+        public CommandMessage Message { get; set; }
+
+        [global::ProtoBuf.ProtoMember(10, Name = @"ack")]
+        public CommandAck Ack { get; set; }
+
+        [global::ProtoBuf.ProtoMember(11, Name = @"flow")]
+        public CommandFlow Flow { get; set; }
+
+        [global::ProtoBuf.ProtoMember(12, Name = @"unsubscribe")]
+        public CommandUnsubscribe Unsubscribe { get; set; }
+
+        [global::ProtoBuf.ProtoMember(13, Name = @"success")]
+        public CommandSuccess Success { get; set; }
+
+        [global::ProtoBuf.ProtoMember(14, Name = @"error")]
+        public CommandError Error { get; set; }
+
+        [global::ProtoBuf.ProtoMember(15, Name = @"close_producer")]
+        public CommandCloseProducer CloseProducer { get; set; }
+
+        [global::ProtoBuf.ProtoMember(16, Name = @"close_consumer")]
+        public CommandCloseConsumer CloseConsumer { get; set; }
+
+        [global::ProtoBuf.ProtoMember(17, Name = @"producer_success")]
+        public CommandProducerSuccess ProducerSuccess { get; set; }
+
+        [global::ProtoBuf.ProtoMember(18, Name = @"ping")]
+        public CommandPing Ping { get; set; }
+
+        [global::ProtoBuf.ProtoMember(19, Name = @"pong")]
+        public CommandPong Pong { get; set; }
+
+        [global::ProtoBuf.ProtoMember(20)]
+        public CommandRedeliverUnacknowledgedMessages RedeliverUnacknowledgedMessages { get; set; }
+
+        [global::ProtoBuf.ProtoMember(21)]
+        public CommandPartitionedTopicMetadata PartitionMetadata { get; set; }
+
+        [global::ProtoBuf.ProtoMember(22)]
+        public CommandPartitionedTopicMetadataResponse PartitionMetadataResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(23)]
+        public CommandLookupTopic LookupTopic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(24)]
+        public CommandLookupTopicResponse LookupTopicResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(25)]
+        public CommandConsumerStats ConsumerStats { get; set; }
+
+        [global::ProtoBuf.ProtoMember(26)]
+        public CommandConsumerStatsResponse ConsumerStatsResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(27)]
+        public CommandReachedEndOfTopic ReachedEndOfTopic { get; set; }
+
+        [global::ProtoBuf.ProtoMember(28, Name = @"seek")]
+        public CommandSeek Seek { get; set; }
+
+        [global::ProtoBuf.ProtoMember(29)]
+        public CommandGetLastMessageId GetLastMessageId { get; set; }
+
+        [global::ProtoBuf.ProtoMember(30)]
+        public CommandGetLastMessageIdResponse GetLastMessageIdResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(31, Name = @"active_consumer_change")]
+        public CommandActiveConsumerChange ActiveConsumerChange { get; set; }
+
+        [global::ProtoBuf.ProtoMember(32)]
+        public CommandGetTopicsOfNamespace GetTopicsOfNamespace { get; set; }
+
+        [global::ProtoBuf.ProtoMember(33)]
+        public CommandGetTopicsOfNamespaceResponse GetTopicsOfNamespaceResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(34)]
+        public CommandGetSchema GetSchema { get; set; }
+
+        [global::ProtoBuf.ProtoMember(35)]
+        public CommandGetSchemaResponse GetSchemaResponse { get; set; }
+
+        [global::ProtoBuf.ProtoMember(36)]
+        public CommandAuthChallenge AuthChallenge { get; set; }
+
+        [global::ProtoBuf.ProtoMember(37)]
+        public CommandAuthResponse AuthResponse { get; set; }
+
+        [global::ProtoBuf.ProtoContract()]
+        public enum Type
+        {
+            [global::ProtoBuf.ProtoEnum(Name = @"CONNECT")]
+            Connect = 2,
+            [global::ProtoBuf.ProtoEnum(Name = @"CONNECTED")]
+            Connected = 3,
+            [global::ProtoBuf.ProtoEnum(Name = @"SUBSCRIBE")]
+            Subscribe = 4,
+            [global::ProtoBuf.ProtoEnum(Name = @"PRODUCER")]
+            Producer = 5,
+            [global::ProtoBuf.ProtoEnum(Name = @"SEND")]
+            Send = 6,
+            [global::ProtoBuf.ProtoEnum(Name = @"SEND_RECEIPT")]
+            SendReceipt = 7,
+            [global::ProtoBuf.ProtoEnum(Name = @"SEND_ERROR")]
+            SendError = 8,
+            [global::ProtoBuf.ProtoEnum(Name = @"MESSAGE")]
+            Message = 9,
+            [global::ProtoBuf.ProtoEnum(Name = @"ACK")]
+            Ack = 10,
+            [global::ProtoBuf.ProtoEnum(Name = @"FLOW")]
+            Flow = 11,
+            [global::ProtoBuf.ProtoEnum(Name = @"UNSUBSCRIBE")]
+            Unsubscribe = 12,
+            [global::ProtoBuf.ProtoEnum(Name = @"SUCCESS")]
+            Success = 13,
+            [global::ProtoBuf.ProtoEnum(Name = @"ERROR")]
+            Error = 14,
+            [global::ProtoBuf.ProtoEnum(Name = @"CLOSE_PRODUCER")]
+            CloseProducer = 15,
+            [global::ProtoBuf.ProtoEnum(Name = @"CLOSE_CONSUMER")]
+            CloseConsumer = 16,
+            [global::ProtoBuf.ProtoEnum(Name = @"PRODUCER_SUCCESS")]
+            ProducerSuccess = 17,
+            [global::ProtoBuf.ProtoEnum(Name = @"PING")]
+            Ping = 18,
+            [global::ProtoBuf.ProtoEnum(Name = @"PONG")]
+            Pong = 19,
+            [global::ProtoBuf.ProtoEnum(Name = @"REDELIVER_UNACKNOWLEDGED_MESSAGES")]
+            RedeliverUnacknowledgedMessages = 20,
+            [global::ProtoBuf.ProtoEnum(Name = @"PARTITIONED_METADATA")]
+            PartitionedMetadata = 21,
+            [global::ProtoBuf.ProtoEnum(Name = @"PARTITIONED_METADATA_RESPONSE")]
+            PartitionedMetadataResponse = 22,
+            [global::ProtoBuf.ProtoEnum(Name = @"LOOKUP")]
+            Lookup = 23,
+            [global::ProtoBuf.ProtoEnum(Name = @"LOOKUP_RESPONSE")]
+            LookupResponse = 24,
+            [global::ProtoBuf.ProtoEnum(Name = @"CONSUMER_STATS")]
+            ConsumerStats = 25,
+            [global::ProtoBuf.ProtoEnum(Name = @"CONSUMER_STATS_RESPONSE")]
+            ConsumerStatsResponse = 26,
+            [global::ProtoBuf.ProtoEnum(Name = @"REACHED_END_OF_TOPIC")]
+            ReachedEndOfTopic = 27,
+            [global::ProtoBuf.ProtoEnum(Name = @"SEEK")]
+            Seek = 28,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_LAST_MESSAGE_ID")]
+            GetLastMessageId = 29,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_LAST_MESSAGE_ID_RESPONSE")]
+            GetLastMessageIdResponse = 30,
+            [global::ProtoBuf.ProtoEnum(Name = @"ACTIVE_CONSUMER_CHANGE")]
+            ActiveConsumerChange = 31,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_TOPICS_OF_NAMESPACE")]
+            GetTopicsOfNamespace = 32,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_TOPICS_OF_NAMESPACE_RESPONSE")]
+            GetTopicsOfNamespaceResponse = 33,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_SCHEMA")]
+            GetSchema = 34,
+            [global::ProtoBuf.ProtoEnum(Name = @"GET_SCHEMA_RESPONSE")]
+            GetSchemaResponse = 35,
+            [global::ProtoBuf.ProtoEnum(Name = @"AUTH_CHALLENGE")]
+            AuthChallenge = 36,
+            [global::ProtoBuf.ProtoEnum(Name = @"AUTH_RESPONSE")]
+            AuthResponse = 37,
+        }
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public enum CompressionType
+    {
+        [global::ProtoBuf.ProtoEnum(Name = @"NONE")]
+        None = 0,
+        [global::ProtoBuf.ProtoEnum(Name = @"LZ4")]
+        Lz4 = 1,
+        [global::ProtoBuf.ProtoEnum(Name = @"ZLIB")]
+        Zlib = 2,
+        [global::ProtoBuf.ProtoEnum(Name = @"ZSTD")]
+        Zstd = 3,
+        [global::ProtoBuf.ProtoEnum(Name = @"SNAPPY")]
+        Snappy = 4,
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public enum ServerError
+    {
+        UnknownError = 0,
+        MetadataError = 1,
+        PersistenceError = 2,
+        AuthenticationError = 3,
+        AuthorizationError = 4,
+        ConsumerBusy = 5,
+        ServiceNotReady = 6,
+        ProducerBlockedQuotaExceededError = 7,
+        ProducerBlockedQuotaExceededException = 8,
+        ChecksumError = 9,
+        UnsupportedVersionError = 10,
+        TopicNotFound = 11,
+        SubscriptionNotFound = 12,
+        ConsumerNotFound = 13,
+        TooManyRequests = 14,
+        TopicTerminatedError = 15,
+        ProducerBusy = 16,
+        InvalidTopicName = 17,
+        IncompatibleSchema = 18,
+        ConsumerAssignError = 19,
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public enum AuthMethod
+    {
+        AuthMethodNone = 0,
+        AuthMethodYcaV1 = 1,
+        AuthMethodAthens = 2,
+    }
+
+    [global::ProtoBuf.ProtoContract()]
+    public enum ProtocolVersion
+    {
+        [global::ProtoBuf.ProtoEnum(Name = @"v0")]
+        V0 = 0,
+        [global::ProtoBuf.ProtoEnum(Name = @"v1")]
+        V1 = 1,
+        [global::ProtoBuf.ProtoEnum(Name = @"v2")]
+        V2 = 2,
+        [global::ProtoBuf.ProtoEnum(Name = @"v3")]
+        V3 = 3,
+        [global::ProtoBuf.ProtoEnum(Name = @"v4")]
+        V4 = 4,
+        [global::ProtoBuf.ProtoEnum(Name = @"v5")]
+        V5 = 5,
+        [global::ProtoBuf.ProtoEnum(Name = @"v6")]
+        V6 = 6,
+        [global::ProtoBuf.ProtoEnum(Name = @"v7")]
+        V7 = 7,
+        [global::ProtoBuf.ProtoEnum(Name = @"v8")]
+        V8 = 8,
+        [global::ProtoBuf.ProtoEnum(Name = @"v9")]
+        V9 = 9,
+        [global::ProtoBuf.ProtoEnum(Name = @"v10")]
+        V10 = 10,
+        [global::ProtoBuf.ProtoEnum(Name = @"v11")]
+        V11 = 11,
+        [global::ProtoBuf.ProtoEnum(Name = @"v12")]
+        V12 = 12,
+        [global::ProtoBuf.ProtoEnum(Name = @"v13")]
+        V13 = 13,
+        [global::ProtoBuf.ProtoEnum(Name = @"v14")]
+        V14 = 14,
+    }
+}
+
+#pragma warning restore CS1591, CS0612, CS3021, IDE1006
\ No newline at end of file
diff --git a/src/DotPulsar/Internal/PulsarClientBuilder.cs b/src/DotPulsar/Internal/PulsarClientBuilder.cs
new file mode 100644
index 0000000..31241e8
--- /dev/null
+++ b/src/DotPulsar/Internal/PulsarClientBuilder.cs
@@ -0,0 +1,35 @@
+﻿using DotPulsar.Abstractions;
+using System;
+using System.Reflection;
+
+namespace DotPulsar.Internal
+{
+    public sealed class PulsarClientBuilder : IPulsarClientBuilder
+    {
+        private static readonly int ProtocolVersion;
+        private static readonly string ClientVersion;
+
+        static PulsarClientBuilder()
+        {
+            ProtocolVersion = 12;
+            var assemblyName = Assembly.GetCallingAssembly().GetName();
+            ClientVersion = assemblyName.Name + " " + assemblyName.Version.ToString(3);
+        }
+
+        private Uri _serviceUrl;
+
+        public PulsarClientBuilder() => _serviceUrl = new Uri("pulsar://localhost:6650");
+
+        public IPulsarClientBuilder ServiceUrl(Uri uri)
+        {
+            _serviceUrl = uri;
+            return this;
+        }
+
+        public IPulsarClient Build()
+        {
+            var connectionPool = new ConnectionPool(ProtocolVersion, ClientVersion, _serviceUrl);
+            return new PulsarClient(connectionPool);
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/PulsarStream.cs b/src/DotPulsar/Internal/PulsarStream.cs
new file mode 100644
index 0000000..ed18b48
--- /dev/null
+++ b/src/DotPulsar/Internal/PulsarStream.cs
@@ -0,0 +1,118 @@
+﻿using DotPulsar.Internal.Extensions;
+using System;
+using System.Buffers;
+using System.IO;
+using System.IO.Pipelines;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class PulsarStream : IDisposable
+    {
+        private readonly Stream _stream;
+        private readonly Action<uint, ReadOnlySequence<byte>> _handler;
+        private readonly CancellationTokenSource _tokenSource;
+
+        public PulsarStream(Stream stream, Action<uint, ReadOnlySequence<byte>> handler)
+        {
+            _stream = stream;
+            _handler = handler;
+            _tokenSource = new CancellationTokenSource();
+            var pipe = new Pipe();
+            var fill = FillPipe(_stream, pipe.Writer, _tokenSource.Token);
+            var read = ReadPipe(pipe.Reader, _tokenSource.Token);
+            IsClosed = Task.WhenAny(fill, read);
+        }
+
+        public Task IsClosed { get; }
+
+        public async Task Send(ReadOnlySequence<byte> sequence)
+        {
+            try
+            {
+                foreach (var segment in sequence)
+                {
+                    await _stream.WriteAsync(segment);
+                }
+            }
+            catch
+            {
+                _tokenSource.Cancel();
+                throw;
+            }
+        }
+
+        public void Dispose()
+        {
+            _tokenSource.Cancel();
+            _stream.Dispose();
+        }
+
+        private async Task FillPipe(Stream stream, PipeWriter writer, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var memory = writer.GetMemory(84999); // LOH - 1 byte
+                    var bytesRead = await stream.ReadAsync(memory, cancellationToken);
+                    if (bytesRead == 0)
+                        break;
+
+                    writer.Advance(bytesRead);
+
+                    var result = await writer.FlushAsync(cancellationToken);
+                    if (result.IsCompleted)
+                        break;
+                }
+            }
+            catch
+            {
+                _tokenSource.Cancel();
+            }
+
+            writer.Complete();
+        }
+
+        private async Task ReadPipe(PipeReader reader, CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    var result = await reader.ReadAsync(cancellationToken);
+                    var buffer = result.Buffer;
+
+                    while (!cancellationToken.IsCancellationRequested)
+                    {
+                        if (buffer.Length < 4)
+                            break;
+
+                        var frameSize = buffer.ReadUInt32(0, true);
+                        var totalSize = frameSize + 4;
+                        if (buffer.Length < totalSize)
+                            break;
+
+                        var commandSize = buffer.ReadUInt32(4, true);
+
+                        _handler(commandSize, buffer.Slice(8, totalSize - 8));
+
+                        buffer = buffer.Slice(totalSize);
+                    }
+
+                    if (result.IsCompleted)
+                        break;
+
+                    reader.AdvanceTo(buffer.Start);
+                }
+            }
+            catch
+            {
+                _tokenSource.Cancel();
+            }
+
+            reader.Complete();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/Reader.cs b/src/DotPulsar/Internal/Reader.cs
new file mode 100644
index 0000000..a4427e0
--- /dev/null
+++ b/src/DotPulsar/Internal/Reader.cs
@@ -0,0 +1,105 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Exceptions;
+using DotPulsar.Internal.Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class Reader : IReader
+    {
+        private readonly Executor _executor;
+        private readonly IConsumerStreamFactory _streamFactory;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly StateManager<ReaderState> _stateManager;
+        private readonly CancellationTokenSource _connectTokenSource;
+        private readonly Task _connectTask;
+        private Action _throwIfClosedOrFaulted;
+        private IConsumerStream Stream { get; set; }
+
+        public Reader(IConsumerStreamFactory streamFactory, IFaultStrategy faultStrategy)
+        {
+            _executor = new Executor(ExecutorOnException);
+            _stateManager = new StateManager<ReaderState>(ReaderState.Disconnected, ReaderState.Closed, ReaderState.ReachedEndOfTopic, ReaderState.Faulted);
+            _streamFactory = streamFactory;
+            _faultStrategy = faultStrategy;
+            _connectTokenSource = new CancellationTokenSource();
+            Stream = new NotReadyStream();
+            _connectTask = Connect(_connectTokenSource.Token);
+            _throwIfClosedOrFaulted = () => { };
+        }
+
+        public async Task<ReaderState> StateChangedTo(ReaderState state, CancellationToken cancellationToken) => await _stateManager.StateChangedTo(state, cancellationToken);
+        public async Task<ReaderState> StateChangedFrom(ReaderState state, CancellationToken cancellationToken) => await _stateManager.StateChangedFrom(state, cancellationToken);
+        public bool IsFinalState() => _stateManager.IsFinalState();
+        public bool IsFinalState(ReaderState state) => _stateManager.IsFinalState(state);
+
+        public void Dispose()
+        {
+            _executor.Dispose();
+            _connectTokenSource.Cancel();
+            _connectTask.Wait();
+        }
+
+        public async Task<Message> Receive(CancellationToken cancellationToken) => await _executor.Execute(() => Stream.Receive(cancellationToken), cancellationToken);
+
+        private async Task ExecutorOnException(Exception exception, CancellationToken cancellationToken)
+        {
+            _throwIfClosedOrFaulted();
+
+            switch (_faultStrategy.DetermineFaultAction(exception))
+            {
+                case FaultAction.Retry:
+                    await Task.Delay(_faultStrategy.TimeToWait, cancellationToken);
+                    break;
+                case FaultAction.Relookup:
+                    await _stateManager.StateChangedTo(ReaderState.Connected, cancellationToken);
+                    break;
+                case FaultAction.Fault:
+                    HasFaulted(exception);
+                    break;
+            }
+
+            _throwIfClosedOrFaulted();
+        }
+
+        private void HasFaulted(Exception exception)
+        {
+            _throwIfClosedOrFaulted = () => throw exception;
+            _stateManager.SetState(ReaderState.Faulted);
+        }
+
+        private void HasClosed()
+        {
+            _throwIfClosedOrFaulted = () => throw new ReaderClosedException();
+            _stateManager.SetState(ReaderState.Closed);
+        }
+
+        private async Task Connect(CancellationToken cancellationToken)
+        {
+            try
+            {
+                while (true)
+                {
+                    using (var proxy = new ReaderProxy(_stateManager, new AsyncQueue<MessagePackage>()))
+                    using (Stream = await _streamFactory.CreateStream(proxy, cancellationToken))
+                    {
+                        proxy.Active();
+                        await _stateManager.StateChangedFrom(ReaderState.Connected, cancellationToken);
+                        if (_stateManager.IsFinalState())
+                            return;
+                    }
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                HasClosed();
+            }
+            catch (Exception exception)
+            {
+                HasFaulted(exception);
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/ReaderBuilder.cs b/src/DotPulsar/Internal/ReaderBuilder.cs
new file mode 100644
index 0000000..85de19d
--- /dev/null
+++ b/src/DotPulsar/Internal/ReaderBuilder.cs
@@ -0,0 +1,42 @@
+﻿using DotPulsar.Abstractions;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ReaderBuilder : IReaderBuilder
+    {
+        private readonly IPulsarClient _pulsarClient;
+        private readonly ReaderOptions _options;
+
+        public ReaderBuilder(IPulsarClient pulsarClient)
+        {
+            _pulsarClient = pulsarClient;
+            _options = new ReaderOptions();
+        }
+
+        public IReaderBuilder ReaderName(string name)
+        {
+            _options.ReaderName = name;
+            return this;
+        }
+
+        public IReaderBuilder MessagePrefetchCount(uint count)
+        {
+            _options.MessagePrefetchCount = count;
+            return this;
+        }
+
+        public IReaderBuilder StartMessageId(MessageId messageId)
+        {
+            _options.StartMessageId = messageId;
+            return this;
+        }
+
+        public IReaderBuilder Topic(string topic)
+        {
+            _options.Topic = topic;
+            return this;
+        }
+
+        public IReader Create() => _pulsarClient.CreateReader(_options);
+    }
+}
diff --git a/src/DotPulsar/Internal/ReaderProxy.cs b/src/DotPulsar/Internal/ReaderProxy.cs
new file mode 100644
index 0000000..eb93582
--- /dev/null
+++ b/src/DotPulsar/Internal/ReaderProxy.cs
@@ -0,0 +1,53 @@
+﻿using DotPulsar.Internal.Abstractions;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class ReaderProxy : IConsumerProxy, IDisposable
+    {
+        private readonly object _lock;
+        private readonly StateManager<ReaderState> _stateManager;
+        private readonly AsyncQueue<MessagePackage> _queue;
+        private bool _hasDisconnected;
+
+        public ReaderProxy(StateManager<ReaderState> stateManager, AsyncQueue<MessagePackage> queue)
+        {
+            _lock = new object();
+            _stateManager = stateManager;
+            _queue = queue;
+            _hasDisconnected = false;
+        }
+
+        public void Active() => SetState(ReaderState.Connected);
+        public void Inactive() => SetState(ReaderState.Connected);
+        public void ReachedEndOfTopic() => SetState(ReaderState.ReachedEndOfTopic);
+
+        public void Disconnected()
+        {
+            lock (_lock)
+            {
+                if (_hasDisconnected)
+                    return;
+
+                _stateManager.SetState(ReaderState.Disconnected);
+                _hasDisconnected = true;
+            }
+        }
+
+        public void Enqueue(MessagePackage package) => _queue.Enqueue(package);
+        public async Task<MessagePackage> Dequeue(CancellationToken cancellationToken) => await _queue.Dequeue(cancellationToken);
+
+        private void SetState(ReaderState state)
+        {
+            lock (_lock)
+            {
+                if (!_hasDisconnected)
+                    _stateManager.SetState(state);
+            }
+        }
+
+        public void Dispose() => _queue.Dispose();
+    }
+}
diff --git a/src/DotPulsar/Internal/RequestResponseHandler.cs b/src/DotPulsar/Internal/RequestResponseHandler.cs
new file mode 100644
index 0000000..8d11e95
--- /dev/null
+++ b/src/DotPulsar/Internal/RequestResponseHandler.cs
@@ -0,0 +1,93 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class RequestResponseHandler : IDisposable
+    {
+        private readonly Awaitor<string, BaseCommand> _responses;
+        private ulong _requestId;
+
+        public RequestResponseHandler()
+        {
+            _responses = new Awaitor<string, BaseCommand>();
+            _requestId = 1;
+        }
+
+        public void Dispose() => _responses.Dispose();
+
+        public Task<BaseCommand> Outgoing(BaseCommand command)
+        {
+            SetRequestId(command);
+            return _responses.CreateTask(GetResponseIdentifier(command));
+        }
+
+        public void Incoming(BaseCommand command)
+        {
+            var identifier = GetResponseIdentifier(command);
+            if (identifier != null)
+                _responses.SetResult(identifier, command);
+        }
+
+        private void SetRequestId(BaseCommand cmd)
+        {
+            switch (cmd.CommandType)
+            {
+                case BaseCommand.Type.Seek:
+                    cmd.Seek.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.Lookup:
+                    cmd.LookupTopic.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.Error:
+                    cmd.Error.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.Producer:
+                    cmd.Producer.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.CloseProducer:
+                    cmd.CloseProducer.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.Subscribe:
+                    cmd.Subscribe.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.Unsubscribe:
+                    cmd.Unsubscribe.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.CloseConsumer:
+                    cmd.CloseConsumer.RequestId = _requestId++;
+                    return;
+                case BaseCommand.Type.GetLastMessageId:
+                    cmd.GetLastMessageId.RequestId = _requestId++;
+                    return;
+            }
+        }
+
+        private static string GetResponseIdentifier(BaseCommand cmd)
+        {
+            switch (cmd.CommandType)
+            {
+                case BaseCommand.Type.Connect:
+                case BaseCommand.Type.Connected: return "Connected";
+                case BaseCommand.Type.Send: return cmd.Send.ProducerId.ToString() + '-' + cmd.Send.SequenceId.ToString();
+                case BaseCommand.Type.SendError: return cmd.SendError.ProducerId.ToString() + '-' + cmd.SendError.SequenceId.ToString();
+                case BaseCommand.Type.SendReceipt: return cmd.SendReceipt.ProducerId.ToString() + '-' + cmd.SendReceipt.SequenceId.ToString();
+                case BaseCommand.Type.Error: return cmd.Error.RequestId.ToString();
+                case BaseCommand.Type.Producer: return cmd.Producer.RequestId.ToString();
+                case BaseCommand.Type.ProducerSuccess: return cmd.ProducerSuccess.RequestId.ToString();
+                case BaseCommand.Type.CloseProducer: return cmd.CloseProducer.RequestId.ToString();
+                case BaseCommand.Type.Lookup: return cmd.LookupTopic.RequestId.ToString();
+                case BaseCommand.Type.LookupResponse: return cmd.LookupTopicResponse.RequestId.ToString();
+                case BaseCommand.Type.Unsubscribe: return cmd.Unsubscribe.RequestId.ToString();
+                case BaseCommand.Type.Subscribe: return cmd.Subscribe.RequestId.ToString();
+                case BaseCommand.Type.Success: return cmd.Success.RequestId.ToString();
+                case BaseCommand.Type.Seek: return cmd.Seek.RequestId.ToString();
+                case BaseCommand.Type.CloseConsumer: return cmd.CloseConsumer.RequestId.ToString();
+                case BaseCommand.Type.GetLastMessageId: return cmd.GetLastMessageId.RequestId.ToString();
+                case BaseCommand.Type.GetLastMessageIdResponse: return cmd.GetLastMessageIdResponse.RequestId.ToString();
+                default: throw new ArgumentOutOfRangeException("CommandType", cmd.CommandType, "CommandType not supported as request/response type");
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/SendPackage.cs b/src/DotPulsar/Internal/SendPackage.cs
new file mode 100644
index 0000000..7d9fbfd
--- /dev/null
+++ b/src/DotPulsar/Internal/SendPackage.cs
@@ -0,0 +1,14 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System;
+
+namespace DotPulsar.Internal
+{
+    public sealed class SendPackage
+    {
+        public SendPackage(CommandSend command) => Command = command;
+
+        public CommandSend Command { get; }
+        public PulsarApi.MessageMetadata Metadata { get; set; }
+        public ReadOnlyMemory<byte> Payload { get; set; }
+    }
+}
diff --git a/src/DotPulsar/Internal/SequenceBuilder.cs b/src/DotPulsar/Internal/SequenceBuilder.cs
new file mode 100644
index 0000000..4b38e90
--- /dev/null
+++ b/src/DotPulsar/Internal/SequenceBuilder.cs
@@ -0,0 +1,63 @@
+﻿using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace DotPulsar.Internal
+{
+    public sealed class SequenceBuilder<T>
+    {
+        private readonly LinkedList<ReadOnlyMemory<T>> _elements;
+
+        public SequenceBuilder() => _elements = new LinkedList<ReadOnlyMemory<T>>();
+
+        public SequenceBuilder<T> Prepend(ReadOnlyMemory<T> memory)
+        {
+            _elements.AddFirst(memory);
+            return this;
+        }
+
+        public SequenceBuilder<T> Append(ReadOnlyMemory<T> memory)
+        {
+            _elements.AddLast(memory);
+            return this;
+        }
+
+        public long Length => _elements.Sum(e => e.Length);
+
+        public ReadOnlySequence<T> Build()
+        {
+            Segment start = null;
+            Segment current = null;
+
+            foreach (var element in _elements)
+            {
+                if (current is null)
+                {
+                    current = new Segment(element);
+                    start = current;
+                }
+                else
+                    current = current.CreateNext(element);
+            }
+
+            return new ReadOnlySequence<T>(start, 0, current, current.Memory.Length);
+        }
+
+        private sealed class Segment : ReadOnlySequenceSegment<T>
+        {
+            public Segment(ReadOnlyMemory<T> memory, long runningIndex = 0)
+            {
+                Memory = memory;
+                RunningIndex = runningIndex;
+            }
+
+            public Segment CreateNext(ReadOnlyMemory<T> memory)
+            {
+                var segment = new Segment(memory, RunningIndex + Memory.Length);
+                Next = segment;
+                return segment;
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/SequenceId.cs b/src/DotPulsar/Internal/SequenceId.cs
new file mode 100644
index 0000000..624f37a
--- /dev/null
+++ b/src/DotPulsar/Internal/SequenceId.cs
@@ -0,0 +1,16 @@
+﻿namespace DotPulsar.Internal
+{
+    public sealed class SequenceId
+    {
+        public SequenceId(ulong initialSequenceId)
+        {
+            Current = initialSequenceId;
+            if (initialSequenceId > 0)
+                Increment();
+        }
+
+        public ulong Current { get; private set; }
+
+        public void Increment() => ++Current;
+    }
+}
diff --git a/src/DotPulsar/Internal/Serializer.cs b/src/DotPulsar/Internal/Serializer.cs
new file mode 100644
index 0000000..3c2d03c
--- /dev/null
+++ b/src/DotPulsar/Internal/Serializer.cs
@@ -0,0 +1,84 @@
+﻿using DotPulsar.Exceptions;
+using DotPulsar.Internal.Extensions;
+using DotPulsar.Internal.PulsarApi;
+using System;
+using System.Buffers;
+using System.IO;
+
+namespace DotPulsar.Internal
+{
+    public static class Serializer
+    {
+        private static readonly byte[] MagicNumber = new byte[] { 0x0e, 0x01 };
+
+        public static T Deserialize<T>(ReadOnlySequence<byte> sequence)
+        {
+            using var ms = new MemoryStream(sequence.ToArray()); //TODO Fix this when protobuf-net start supporting sequences or .NET supports creating a stream from a sequence
+            return ProtoBuf.Serializer.Deserialize<T>(ms);
+        }
+
+        public static Message Deserialize(MessagePackage package)
+        {
+            var sequence = package.Data;
+            var magicNumberMatches = sequence.StartsWith(MagicNumber);
+            if (!magicNumberMatches)
+                throw new ChecksumException("Magic number don't match");
+            var expectedChecksum = sequence.ReadUInt32(2, true);
+            var actualChecksum = Crc32C.Calculate(sequence.Slice(6));
+            if (expectedChecksum != actualChecksum)
+                throw new ChecksumException(expectedChecksum, actualChecksum);
+            var metaSize = sequence.ReadUInt32(6, true);
+            var meta = Deserialize<PulsarApi.MessageMetadata>(sequence.Slice(10, metaSize));
+            var data = sequence.Slice(10 + metaSize);
+            return new Message(new MessageId(package.Command.MessageId), meta, data);
+        }
+
+        public static ReadOnlySequence<byte> Serialize(BaseCommand command)
+        {
+            var commandBytes = Serialize<BaseCommand>(command);
+            var commandSizeBytes = ToBigEndianBytes((uint)commandBytes.Length);
+            var totalSizeBytes = ToBigEndianBytes((uint)commandBytes.Length + 4);
+
+            return new SequenceBuilder<byte>()
+                .Append(totalSizeBytes)
+                .Append(commandSizeBytes)
+                .Append(commandBytes)
+                .Build();
+        }
+
+        public static ReadOnlySequence<byte> Serialize(BaseCommand command, PulsarApi.MessageMetadata metadata, ReadOnlyMemory<byte> payload)
+        {
+            var commandBytes = Serialize<BaseCommand>(command);
+            var commandSizeBytes = ToBigEndianBytes((uint)commandBytes.Length);
+
+            var metadataBytes = Serialize(metadata);
+            var metadataSizeBytes = ToBigEndianBytes((uint)metadataBytes.Length);
+
+            var sb = new SequenceBuilder<byte>().Append(metadataSizeBytes).Append(metadataBytes).Append(payload);
+            var checksum = Crc32C.Calculate(sb.Build());
+
+            return sb.Prepend(ToBigEndianBytes(checksum))
+                .Prepend(MagicNumber)
+                .Prepend(commandBytes)
+                .Prepend(commandSizeBytes)
+                .Prepend(ToBigEndianBytes((uint)sb.Length))
+                .Build();
+        }
+
+        public static byte[] ToBigEndianBytes(uint integer)
+        {
+            var union = new UIntUnion(integer);
+            if (BitConverter.IsLittleEndian)
+                return new[] { union.B3, union.B2, union.B1, union.B0 };
+            else
+                return new[] { union.B0, union.B1, union.B2, union.B3 };
+        }
+
+        private static byte[] Serialize<T>(T item)
+        {
+            using var ms = new MemoryStream();
+            ProtoBuf.Serializer.Serialize(ms, item);
+            return ms.ToArray();
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/StateChanged.cs b/src/DotPulsar/Internal/StateChanged.cs
new file mode 100644
index 0000000..3690f44
--- /dev/null
+++ b/src/DotPulsar/Internal/StateChanged.cs
@@ -0,0 +1,8 @@
+﻿namespace DotPulsar.Internal
+{
+    public enum StateChanged : byte
+    {
+        To,
+        From
+    }
+}
diff --git a/src/DotPulsar/Internal/StateManager.cs b/src/DotPulsar/Internal/StateManager.cs
new file mode 100644
index 0000000..84557df
--- /dev/null
+++ b/src/DotPulsar/Internal/StateManager.cs
@@ -0,0 +1,75 @@
+﻿using DotPulsar.Abstractions;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class StateManager<TState> : IStateChanged<TState>
+    {
+        private readonly object _lock;
+        private readonly StateTaskCollection<TState> _stateTasks;
+        private readonly TState[] _finalStates;
+
+        public StateManager(TState initialState, params TState[] finalStates)
+        {
+            _lock = new object();
+            _stateTasks = new StateTaskCollection<TState>();
+            _finalStates = finalStates;
+            CurrentState = initialState;
+        }
+
+        public TState CurrentState { get; private set; }
+
+        public TState SetState(TState state)
+        {
+            lock (_lock)
+            {
+                if (IsFinalState(CurrentState) || CurrentState.Equals(state))
+                    return CurrentState;
+
+                var formerState = CurrentState;
+                CurrentState = state;
+
+                if (IsFinalState(CurrentState))
+                    _stateTasks.CompleteAllTasks(CurrentState);
+                else
+                    _stateTasks.CompleteTasksAwaiting(CurrentState);
+
+                return formerState;
+            }
+        }
+
+        public Task<TState> StateChangedTo(TState state, CancellationToken cancellationToken)
+        {
+            lock (_lock)
+            {
+                if (IsFinalState(CurrentState) || CurrentState.Equals(state))
+                    return Task.FromResult(CurrentState);
+                return _stateTasks.CreateTaskFor(state, StateChanged.To, cancellationToken);
+            }
+        }
+
+        public Task<TState> StateChangedFrom(TState state, CancellationToken cancellationToken)
+        {
+            lock (_lock)
+            {
+                if (IsFinalState(CurrentState) || !CurrentState.Equals(state))
+                    return Task.FromResult(CurrentState);
+                return _stateTasks.CreateTaskFor(state, StateChanged.From, cancellationToken);
+            }
+        }
+
+        public bool IsFinalState(TState state)
+        {
+            for (var i = 0; i < _finalStates.Length; ++i)
+            {
+                if (_finalStates[i].Equals(state))
+                    return true;
+            }
+
+            return false;
+        }
+
+        public bool IsFinalState() => IsFinalState(CurrentState);
+    }
+}
diff --git a/src/DotPulsar/Internal/StateTask.cs b/src/DotPulsar/Internal/StateTask.cs
new file mode 100644
index 0000000..aaa17be
--- /dev/null
+++ b/src/DotPulsar/Internal/StateTask.cs
@@ -0,0 +1,29 @@
+﻿using System;
+
+namespace DotPulsar.Internal
+{
+    public sealed class StateTask<TState> : IDisposable
+    {
+        private readonly TState _state;
+        private readonly StateChanged _change;
+
+        public StateTask(TState state, StateChanged change)
+        {
+            _state = state;
+            _change = change;
+            CancelableCompletionSource = new CancelableCompletionSource<TState>();
+        }
+
+        public CancelableCompletionSource<TState> CancelableCompletionSource { get; }
+
+        public void Dispose() => CancelableCompletionSource.Dispose();
+
+        public bool IsAwaiting(TState state)
+        {
+            if (_change == StateChanged.To)
+                return _state.Equals(state);
+
+            return !_state.Equals(state);
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/StateTaskCollection.cs b/src/DotPulsar/Internal/StateTaskCollection.cs
new file mode 100644
index 0000000..eaea664
--- /dev/null
+++ b/src/DotPulsar/Internal/StateTaskCollection.cs
@@ -0,0 +1,79 @@
+﻿using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace DotPulsar.Internal
+{
+    public sealed class StateTaskCollection<TState>
+    {
+        private readonly object _lock;
+        private readonly LinkedList<StateTask<TState>> _awaitors;
+
+        public StateTaskCollection()
+        {
+            _lock = new object();
+            _awaitors = new LinkedList<StateTask<TState>>();
+        }
+
+        public Task<TState> CreateTaskFor(TState state, StateChanged changed, CancellationToken cancellationToken)
+        {
+            LinkedListNode<StateTask<TState>> node;
+
+            lock (_lock)
+            {
+                node = _awaitors.AddFirst(new StateTask<TState>(state, changed));
+            }
+
+            node.Value.CancelableCompletionSource.SetupCancellation(() => TaskWasCanceled(node), cancellationToken);
+            return node.Value.CancelableCompletionSource.Task;
+        }
+
+        public void CompleteTasksAwaiting(TState state)
+        {
+            lock (_lock)
+            {
+                var awaitor = _awaitors.First;
+                while (awaitor != null)
+                {
+                    var next = awaitor.Next;
+                    if (awaitor.Value.IsAwaiting(state))
+                    {
+                        _awaitors.Remove(awaitor);
+                        awaitor.Value.CancelableCompletionSource.SetResult(state);
+                        awaitor.Value.CancelableCompletionSource.Dispose();
+                    }
+                    awaitor = next;
+                }
+            }
+        }
+
+        public void CompleteAllTasks(TState state)
+        {
+            lock (_lock)
+            {
+                foreach (var awaitor in _awaitors)
+                {
+                    awaitor.CancelableCompletionSource.SetResult(state);
+                    awaitor.CancelableCompletionSource.Dispose();
+                }
+                _awaitors.Clear();
+            }
+        }
+
+        private void TaskWasCanceled(LinkedListNode<StateTask<TState>> node)
+        {
+            lock (_lock)
+            {
+                try
+                {
+                    _awaitors.Remove(node);
+                    node.Value.Dispose();
+                }
+                catch
+                {
+                    // Ignore
+                }
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/Internal/SubscribeResponse.cs b/src/DotPulsar/Internal/SubscribeResponse.cs
new file mode 100644
index 0000000..a401525
--- /dev/null
+++ b/src/DotPulsar/Internal/SubscribeResponse.cs
@@ -0,0 +1,9 @@
+﻿namespace DotPulsar.Internal
+{
+    public sealed class SubscribeResponse
+    {
+        public SubscribeResponse(ulong consumerId) => ConsumerId = consumerId;
+
+        public ulong ConsumerId { get; }
+    }
+}
diff --git a/src/DotPulsar/Internal/UIntUnion.cs b/src/DotPulsar/Internal/UIntUnion.cs
new file mode 100644
index 0000000..160bddd
--- /dev/null
+++ b/src/DotPulsar/Internal/UIntUnion.cs
@@ -0,0 +1,38 @@
+﻿using System.Runtime.InteropServices;
+
+namespace DotPulsar.Internal
+{
+    [StructLayout(LayoutKind.Explicit)]
+    public struct UIntUnion
+    {
+        public UIntUnion(byte b0, byte b1, byte b2, byte b3)
+        {
+            UInt = 0;
+            B0 = b0;
+            B1 = b1;
+            B2 = b2;
+            B3 = b3;
+        }
+
+        public UIntUnion(uint value)
+        {
+            B0 = 0;
+            B1 = 0;
+            B2 = 0;
+            B3 = 0;
+            UInt = value;
+        }
+
+        [FieldOffset(0)]
+        public byte B0;
+        [FieldOffset(1)]
+        public byte B1;
+        [FieldOffset(2)]
+        public byte B2;
+        [FieldOffset(3)]
+        public byte B3;
+
+        [FieldOffset(0)]
+        public uint UInt;
+    }
+}
diff --git a/src/DotPulsar/Message.cs b/src/DotPulsar/Message.cs
new file mode 100644
index 0000000..eb553b5
--- /dev/null
+++ b/src/DotPulsar/Message.cs
@@ -0,0 +1,35 @@
+﻿using System.Buffers;
+using System.Collections.Immutable;
+
+namespace DotPulsar
+{
+    public sealed class Message
+    {
+        private readonly Internal.PulsarApi.MessageMetadata _messageMetadata;
+        private ImmutableDictionary<string, string> _properties;
+
+        internal Message(MessageId messageId, Internal.PulsarApi.MessageMetadata messageMetadata, ReadOnlySequence<byte> data)
+        {
+            MessageId = messageId;
+            _messageMetadata = messageMetadata;
+            Data = data;
+        }
+
+        public MessageId MessageId { get; }
+        public ReadOnlySequence<byte> Data { get; }
+        public ulong EventTime => _messageMetadata.EventTime;
+        public string ProducerName => _messageMetadata.ProducerName;
+        public ulong PublishTime => _messageMetadata.PublishTime;
+        public ulong SequenceId => _messageMetadata.SequenceId;
+
+        public ImmutableDictionary<string, string> Properties
+        {
+            get
+            {
+                if (_properties == null)
+                    _properties = _messageMetadata.Properties.ToImmutableDictionary(p => p.Key, p => p.Value);
+                return _properties;
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/MessageId.cs b/src/DotPulsar/MessageId.cs
new file mode 100644
index 0000000..1c163af
--- /dev/null
+++ b/src/DotPulsar/MessageId.cs
@@ -0,0 +1,46 @@
+﻿using DotPulsar.Internal.PulsarApi;
+using System;
+
+namespace DotPulsar
+{
+    public sealed class MessageId : IEquatable<MessageId>
+    {
+        static MessageId()
+        {
+            Earliest = new MessageId(ulong.MaxValue, ulong.MaxValue, -1, -1);
+            Latest = new MessageId(long.MaxValue, long.MaxValue, -1, -1);
+        }
+
+        public static MessageId Earliest { get; }
+        public static MessageId Latest { get; }
+
+        internal MessageId(MessageIdData messageIdData) => Data = messageIdData;
+
+        public MessageId(ulong ledgerId, ulong entryId, int partition, int batchIndex)
+        {
+            Data = new MessageIdData
+            {
+                LedgerId = ledgerId,
+                EntryId = entryId,
+                Partition = partition,
+                BatchIndex = batchIndex
+            };
+        }
+
+        internal MessageIdData Data { get; }
+
+        public ulong LedgerId => Data.LedgerId;
+        public ulong EntryId => Data.EntryId;
+        public int Partition => Data.Partition;
+        public int BatchIndex => Data.BatchIndex;
+
+        public override bool Equals(object o) => o is MessageId ? Equals((MessageId)o) : false;
+        public bool Equals(MessageId other) => LedgerId == other.LedgerId && EntryId == other.EntryId && Partition == other.Partition && BatchIndex == other.BatchIndex;
+
+        public static bool operator ==(MessageId x, MessageId y) => x.Equals(y);
+        public static bool operator !=(MessageId x, MessageId y) => !x.Equals(y);
+
+        public override int GetHashCode() => HashCode.Combine(LedgerId, EntryId, Partition, BatchIndex);
+        public override string ToString() => $"LedgerId: {LedgerId}, EntryId: {EntryId}, Partition: {Partition}, BatchIndex: {BatchIndex}";
+    }
+}
diff --git a/src/DotPulsar/MessageMetadata.cs b/src/DotPulsar/MessageMetadata.cs
new file mode 100644
index 0000000..006abd2
--- /dev/null
+++ b/src/DotPulsar/MessageMetadata.cs
@@ -0,0 +1,49 @@
+﻿namespace DotPulsar
+{
+    public sealed class MessageMetadata
+    {
+        public MessageMetadata() => Metadata = new Internal.PulsarApi.MessageMetadata();
+
+        internal Internal.PulsarApi.MessageMetadata Metadata;
+
+        public ulong EventTime
+        {
+            get => Metadata.EventTime;
+            set => Metadata.EventTime = value;
+        }
+
+        public string this[string key]
+        {
+            get
+            {
+                for (var i = 0; i < Metadata.Properties.Count; ++i)
+                {
+                    var keyValye = Metadata.Properties[i];
+                    if (keyValye.Key == key)
+                        return keyValye.Value;
+                }
+
+                return null;
+            }
+            set
+            {
+                for (var i = 0; i < Metadata.Properties.Count; ++i)
+                {
+                    var keyValye = Metadata.Properties[i];
+                    if (keyValye.Key != key)
+                        continue;
+                    keyValye.Value = value;
+                    return;
+                }
+
+                Metadata.Properties.Add(new Internal.PulsarApi.KeyValue { Key = key, Value = value });
+            }
+        }
+
+        public ulong SequenceId
+        {
+            get => Metadata.SequenceId;
+            set => Metadata.SequenceId = value;
+        }
+    }
+}
diff --git a/src/DotPulsar/ProducerOptions.cs b/src/DotPulsar/ProducerOptions.cs
new file mode 100644
index 0000000..c161b11
--- /dev/null
+++ b/src/DotPulsar/ProducerOptions.cs
@@ -0,0 +1,9 @@
+﻿namespace DotPulsar
+{
+    public sealed class ProducerOptions
+    {
+        public string ProducerName { get; set; }
+        public ulong InitialSequenceId { get; set; }
+        public string Topic { get; set; }
+    }
+}
diff --git a/src/DotPulsar/ProducerState.cs b/src/DotPulsar/ProducerState.cs
new file mode 100644
index 0000000..0326e88
--- /dev/null
+++ b/src/DotPulsar/ProducerState.cs
@@ -0,0 +1,10 @@
+﻿namespace DotPulsar
+{
+    public enum ProducerState : byte
+    {
+        Closed,
+        Connected,
+        Disconnected,
+        Faulted
+    }
+}
diff --git a/src/DotPulsar/PulsarClient.cs b/src/DotPulsar/PulsarClient.cs
new file mode 100644
index 0000000..7fa2219
--- /dev/null
+++ b/src/DotPulsar/PulsarClient.cs
@@ -0,0 +1,94 @@
+﻿using DotPulsar.Abstractions;
+using DotPulsar.Exceptions;
+using DotPulsar.Internal;
+using DotPulsar.Internal.Abstractions;
+using System;
+using System.Collections.Generic;
+
+namespace DotPulsar
+{
+    public sealed class PulsarClient : IPulsarClient
+    {
+        private readonly object _lock;
+        private readonly IFaultStrategy _faultStrategy;
+        private readonly LinkedList<IDisposable> _disposabels;
+        private readonly ConnectionPool _connectionPool;
+        private bool _isClosed;
+
+        internal PulsarClient(ConnectionPool connectionPool)
+        {
+            _lock = new object();
+            _faultStrategy = new FaultStrategy(3000);
+            _disposabels = new LinkedList<IDisposable>();
+            _connectionPool = connectionPool;
+            _isClosed = false;
+        }
+
+        public static IPulsarClientBuilder Builder() => new PulsarClientBuilder();
+
+        public IProducer CreateProducer(ProducerOptions options)
+        {
+            lock (_lock)
+            {
+                ThrowIfClosed();
+                var producer = new Producer(new ProducerStreamFactory(_connectionPool, options, _faultStrategy), _faultStrategy);
+                _disposabels.AddFirst(producer);
+                producer.StateChangedTo(ProducerState.Closed, default).ContinueWith(t => Remove(producer));
+                return producer;
+            }
+        }
+
+        public IConsumer CreateConsumer(ConsumerOptions options)
+        {
+            lock (_lock)
+            {
+                ThrowIfClosed();
+                var consumer = new Consumer(new ConsumerStreamFactory(_connectionPool, options, _faultStrategy), _faultStrategy, options.SubscriptionType != SubscriptionType.Failover);
+                _disposabels.AddFirst(consumer);
+                consumer.StateChangedTo(ConsumerState.Closed, default).ContinueWith(t => Remove(consumer));
+                return consumer;
+            }
+        }
+
+        public IReader CreateReader(ReaderOptions options)
+        {
+            lock (_lock)
+            {
+                ThrowIfClosed();
+                var reader = new Reader(new ConsumerStreamFactory(_connectionPool, options, _faultStrategy), _faultStrategy);
+                _disposabels.AddFirst(reader);
+                reader.StateChangedTo(ReaderState.Closed, default).ContinueWith(t => Remove(reader));
+                return reader;
+            }
+        }
+
+        public void Dispose() //While we wait for IAsyncDisposable
+        {
+            lock (_lock)
+            {
+                ThrowIfClosed();
+                _isClosed = true;
+                foreach (var disposable in _disposabels)
+                {
+                    disposable.Dispose();
+                }
+            }
+
+            _connectionPool.Dispose();
+        }
+
+        private void ThrowIfClosed()
+        {
+            if (_isClosed)
+                throw new PulsarClientClosedException();
+        }
+
+        private void Remove(IDisposable disposable)
+        {
+            lock (_lock)
+            {
+                _disposabels.Remove(disposable);
+            }
+        }
+    }
+}
diff --git a/src/DotPulsar/ReaderOptions.cs b/src/DotPulsar/ReaderOptions.cs
new file mode 100644
index 0000000..a5087d1
--- /dev/null
+++ b/src/DotPulsar/ReaderOptions.cs
@@ -0,0 +1,10 @@
+﻿namespace DotPulsar
+{
+    public sealed class ReaderOptions
+    {
+        public string ReaderName { get; set; }
+        public uint MessagePrefetchCount { get; set; }
+        public MessageId StartMessageId { get; set; }
+        public string Topic { get; set; }
+    }
+}
diff --git a/src/DotPulsar/ReaderState.cs b/src/DotPulsar/ReaderState.cs
new file mode 100644
index 0000000..32654e6
--- /dev/null
+++ b/src/DotPulsar/ReaderState.cs
@@ -0,0 +1,11 @@
+﻿namespace DotPulsar
+{
+    public enum ReaderState : byte
+    {
+        Closed,
+        Connected,
+        Disconnected,
+        Faulted,
+        ReachedEndOfTopic
+    }
+}
diff --git a/src/DotPulsar/SubscriptionInitialPosition.cs b/src/DotPulsar/SubscriptionInitialPosition.cs
new file mode 100644
index 0000000..8c352d0
--- /dev/null
+++ b/src/DotPulsar/SubscriptionInitialPosition.cs
@@ -0,0 +1,8 @@
+﻿namespace DotPulsar
+{
+    public enum SubscriptionInitialPosition : byte
+    {
+        Latest = 0,
+        Earliest = 1
+    }
+}
diff --git a/src/DotPulsar/SubscriptionType.cs b/src/DotPulsar/SubscriptionType.cs
new file mode 100644
index 0000000..aa815ce
--- /dev/null
+++ b/src/DotPulsar/SubscriptionType.cs
@@ -0,0 +1,9 @@
+﻿namespace DotPulsar
+{
+    public enum SubscriptionType : byte
+    {
+        Exclusive = 0,
+        Shared = 1,
+        Failover = 2
+    }
+}