How valid data leads to problematic conclusions

For people who got used to the New Project dialog, finding an appropriate project type raises no issues at all. The tree-like structure (the same for years) make the navigation quick and you always follow the same pattern "language-framework-project type".

Visual Studio 2019 introduces the new look of the dialog. There's a Visual Studio Blog entry with a story behind the new look. In short, they've collected the telemetry data (the data that is automatically collected by the VS and sent back to Microsoft), performed some interviews and observed that developers are confused by the taxonomy presented on the project tree which causes the drop-off to opening a project.

Where the telemetry data is possibly objective, the interpretation is not. The team seems to put everyone in the very same basket - someone who just downloaded the VS and runs it for the first time in their life and a professional who uses VS for almost two decades. In the latter case, the team observed that even for experienced developers, some of the options weren’t intuitive. Most couldn’t pinpoint the link to ‘Open Visual Studio Installer’ when the templates they were looking for weren’t available. What's the solution to all these (very different issues)?

This:

Every single developer I ask (myself included) is disappointed by the new look and the behavior of this something. Example comments from below the blog entry:

• From my own experience, I find it more confusing and time consuming to create new projects with the new dialog concept.
• Yes, this new dialog is the worst experience from VS 2019.
• I liked old new project dialog more. New one is confusing.
• This new dialog just ensures that most developers will only ever see the top 5 or so project templates and that the rest of them may as well not even exist.
• I absolutely positvely hate the new dialog.

The list continues in a similar manner so guess what ... the comment section has been closed.

The good news is that there's an Extension in the marketplace, the Classis "New Project" Dialog and the feedback here is expected

• Thank you. Thank you. Thank you.
• From one extension developer to another one - well done!
• Thank you! This saved my life. I found the out of box "New project" dialog of VS2019 to be very limiting.

Download and rate this extension and have at least one inconvenience resolved. Whoever wrote the extension - thank you for your contribution.

Your XSLT processor is Turing Complete

Just by change I stumpled upon a nice approach to proving that XSLT processors are Turing Complete. This approach by Unidex involves a XSLT definition that implements the Turing Machine and an example language of writing Turing machines in XML.

(Note that the page is there for quite some time so when it's down somewhere in future, use the WayBackMachine to access it)

If you want to try it, download the XSLT, download the example and put the two together with a simple snippet:

        static void Main(string[] args)
        {
            var transform = new XslCompiledTransform();
            transform.Load(XmlReader.Create(File.Open("turing.xslt", FileMode.Open)));

            StringBuilder sb = new StringBuilder();
            StringWriter sw = new StringWriter(sb);

            var p = new XsltArgumentList();
            p.AddParam("tape", "", "199");
            transform.Transform(XmlReader.Create(File.Open("turing.xml", FileMode.Open)), p, sw);

            Console.WriteLine(sb.ToString());

            Console.ReadLine();
        }

Running this yields

Step number = 1
Tape        = 199
Tape head     ^
State       = go_right
Next symbol = 1
Next state  = go_right
Movement    = right

Step number = 2
Tape        = 199
Tape head      ^
State       = go_right
Next symbol = 9
Next state  = go_right
Movement    = right

Step number = 3
Tape        = 199
Tape head       ^
State       = go_right
Next symbol = 9
Next state  = go_right
Movement    = right

Step number = 4
Tape        = 199
Tape head        ^
State       = go_right
Next symbol =
Next state  = increment
Movement    = left

Step number = 5
Tape        = 199
Tape head       ^
State       = increment
Next symbol = 0
Next state  = increment
Movement    = left

Step number = 6
Tape        = 190
Tape head      ^
State       = increment
Next symbol = 0
Next state  = increment
Movement    = left

Step number = 7
Tape        = 100
Tape head     ^
State       = increment
Next symbol = 2
Next state  = stop
Movement    = left

The Turing machine has halted.
Final state = stop
Final tape  =  200
Tape head     ^

Tetris challenge - reloaded (Javascript)

Years ago I've blogged on a challenge which was to implement a simple tetris game. This new reloaded version is a direct one-to-one translation to vanilla Javascript and runs in a browser. Feel free to learn from the code and/or reuse it in any way. Note that the UP key has to be pressed to start the game.

(Re)start: up
Moves: left/right/down/space

The code:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="ie=edge">
    <title>Document</title>

    <script>

        var pieceMove = {
            left:   'left',
            right:  'right',
            up:     'up',
            down:   'down',
            rotate: 'rotate'
        };
    
        function board() {
            this.MARGIN      = 5;
            this.SIZEX       = 10;
            this.SIZEY       = 40;
            this.BLOCKSIZE   = 10;
            this.PIECESTARTX = 8;
    
            var i, j;
    
            this.squares = new Array( this.MARGIN + this.SIZEX + this.MARGIN );
            for ( i=0; i < this.squares.length; i++ )
                this.squares[i] = new Array( this.MARGIN + this.SIZEY + this.MARGIN );
    
            // fill the margin 
            for ( i=0; i < this.MARGIN + this.SIZEX + this.MARGIN; i++ )
                for ( j=0; j < this.MARGIN; j++ )
                    this.squares[i][j] = this.squares[i][j + this.SIZEY + this.MARGIN] = true;
            for ( i=0; i < this.MARGIN; i++ )
                for ( j=0; j < this.MARGIN + this.SIZEY + this.MARGIN; j++ )
                    this.squares[i][j] = this.squares[i + this.SIZEX + this.MARGIN][j] = true;        
    
            this.currentPiece  = piece.getRandomPiece();
            this.currentPieceX = this.PIECESTARTX;
            this.currentPieceY = this.MARGIN;        
    
            this.score         = 0;
        }
    
        board.prototype.obtainContext = function(canvas) {
            this.context       = canvas.getContext('2d');     
            this.contextWidth  = canvas.width;
            this.contextHeight = canvas.height;       
        }
    
        board.prototype.paint = function() {
            
            var i, j;
    
            this.context.fillStyle = 'white';
            this.context.fillRect(0, 0, this.contextWidth, this.contextHeight);
    
            // background
            this.context.fillStyle = 'black';
    
            for ( i=0; i < this.SIZEX; i++ )
                for ( j=0; j < this.SIZEY; j++ )
                    if ( this.squares[this.MARGIN + i][this.MARGIN + j] )
                        this.context.fillRect( i * this.BLOCKSIZE, j * this.BLOCKSIZE, this.BLOCKSIZE, this.BLOCKSIZE );
    
            // piece
            if ( this.currentPiece ) {
                for ( i=0; i < this.currentPiece.SIZE; i++ )
                    for ( j=0; j < this.currentPiece.SIZE; j++ )
                        if ( this.currentPiece.blockData[i][j] ) {
                            this.context.fillRect( (this.currentPieceX - this.MARGIN + i) * this.BLOCKSIZE, ( this.currentPieceY - this.MARGIN + j ) * this.BLOCKSIZE, 
                                                    this.BLOCKSIZE, this.BLOCKSIZE);
                        }
            }    

            // score
            this.context.fillStyle = 'red';
            this.context.fillText( this.score, 5, 10 );
        }
    
        /* Advance one step of the game */
        board.prototype.autoUpdate = function() {

            this.removeFullLines();
    
            if ( this.currentPiece ) {
                var _newX = this.currentPieceX;
                var _newY = this.currentPieceY + 1;
    
                if ( this.isLegalPosition( this.currentPiece, _newX, _newY ) ) {
                    this.currentPieceY = _newY;
                }
                else {
                    this.absorbCurrentPiece();
    
                    this.currentPiece  = piece.getRandomPiece();
                    this.currentPieceX = this.PIECESTARTX;
                    this.currentPieceY = this.MARGIN;
    
                    if ( !this.isLegalPosition( this.currentPiece, this.currentPieceX, this.currentPieceY ) ) {
                        this.score = 'Game over: ' + this.score;
                        this.currentPiece = null;

                        return false;
                    }
                }
            }    

            return true;
        }
    
        /* Remove all full lines */
        board.prototype.removeFullLines = function() {
            var fullLineNumber = -1;
            do
            {
                fullLineNumber = this.getPossibleFullLine();
                if ( fullLineNumber >= 0 )
                    this.removeLine( fullLineNumber );
            } while ( fullLineNumber >= 0 );
        }
    
        /* Check if one of lines is possibly filled */
        board.prototype.getPossibleFullLine = function() {
            for ( var j = 0; j < this.SIZEY; j++ )
            {
                var isFull = true;
                for ( var i = 0; i < this.SIZEX; i++ )
                    isFull &= this.squares[this.MARGIN + i][this.MARGIN + j];
    
                if ( isFull )
                    return j + this.MARGIN;
            }
    
            return -1;
        }
    
        /* Remove selected line */
        board.prototype.removeLine = function( lineNumber ) {
            for ( var j = lineNumber; j > this.MARGIN; j-- )
                for ( var i=this.MARGIN; i < this.SIZEX + this.MARGIN; i++ )
                    this.squares[i][j] = this.squares[i][j - 1];
    
            this.score += 1;
        }
    
        /* Copy piece data to board's block data. Called only when piece becomes "solid", non-movable. */
        board.prototype.absorbCurrentPiece = function() {
            if ( this.currentPiece )
                for ( var i=0; i < this.currentPiece.SIZE; i++ )
                    for ( var j=0; j < this.currentPiece.SIZE; j++ )
                        if ( this.currentPiece.blockData[i][j] )
                            this.squares[i + this.currentPieceX][j + this.currentPieceY] = true;
        }
    
        /* Moves current piece into a new position, if possible */
        board.prototype.moveCurrentPiece = function(move) {
    
            var _newPiece = this.currentPiece;
            var _newX     = this.currentPieceX;
            var _newY     = this.currentPieceY;
    
            if ( this.currentPiece )
            {
                if ( move == pieceMove.rotate ) _newPiece = piece.rotate( this.currentPiece );
                if ( move == pieceMove.left )   _newX = this.currentPieceX - 1;
                if ( move == pieceMove.right )  _newX = this.currentPieceX + 1;
                if ( move == pieceMove.down )   _newY = this.currentPieceY + 1;
    
                // move only if legal position
                if ( this.isLegalPosition( _newPiece, _newX, _newY ) )
                {
                    this.currentPiece  = _newPiece;
                    this.currentPieceX = _newX;
                    this.currentPieceY = _newY;
                }
            }
        }
    
        /* Is the new possible position of a piece "legal" in a sense that the piece doesn't overlap with non-empty cells */
        board.prototype.isLegalPosition = function( piece, pieceX, pieceY ) {
            for ( var i=0; i < piece.SIZE; i++ )
                for ( var j=0; j < piece.SIZE; j++ )
                    if ( piece.blockData[i][j] &&
                         this.squares[i + pieceX][j + pieceY]
                        )
                        return false;
    
                return true;
        }
    
        function piece(r0, r1, r2, r3, r4) {
            this.SIZE = 5;
    
            var i;
    
            this.blockData = new Array( this.SIZE );
            for ( i=0; i < this.SIZE; i++ )
                this.blockData[i] = new Array( this.SIZE );
    
            for ( i=0; i<this.SIZE; i++ ) {
                this.blockData[i][0] = r0[i];
                this.blockData[i][1] = r1[i];
                this.blockData[i][2] = r2[i];
                this.blockData[i][3] = r3[i];
                this.blockData[i][4] = r4[i];
            }
        }
    
        piece.rotate = function( source ) {
            if ( source == piece.i_horiz ) return piece.i_vert;
            if ( source == piece.i_vert )  return piece.i_horiz;
    
            if ( source == piece.z_lefthoriz ) return piece.z_leftvert;
            if ( source == piece.z_leftvert )  return piece.z_lefthoriz;
    
            if ( source == piece.z_righthoriz ) return piece.z_rightvert;
            if ( source == piece.z_rightvert )  return piece.z_righthoriz;
    
            if ( source == piece.l_l1 ) return piece.l_l2;
            if ( source == piece.l_l2 ) return piece.l_l3;
            if ( source == piece.l_l3 ) return piece.l_l4;
            if ( source == piece.l_l4 ) return piece.l_l1;
    
            if ( source == piece.r_l1 ) return piece.r_l2;
            if ( source == piece.r_l2 ) return piece.r_l3;
            if ( source == piece.r_l3 ) return piece.r_l4;
            if ( source == piece.r_l4 ) return piece.r_l1;
    
            return piece.box;
        }
    
        piece.getRandomPiece = function() {
            var _next = Math.floor( Math.random() * 6 );
            switch ( _next ) {
                case 1  : return piece.z_lefthoriz;
                case 2  : return piece.z_righthoriz;
                case 3  : return piece.l_l1;
                case 4  : return piece.r_l1;
                case 5  : return piece.i_horiz;
                default : return piece.box;
            }
        }
    
        piece.box = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.i_horiz = new piece(
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.i_vert = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 1, 1, 1, 1 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.z_lefthoriz = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 0, 1, 1, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.z_leftvert = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 1, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.z_righthoriz = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 1, 1, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.z_rightvert = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 1, 0 ],
            [0, 0, 0, 1, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.l_l1 = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 1, 0, 0, 0 ],
            [0, 1, 1, 1, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.l_l2 = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 1, 0, 0, 0 ],
            [0, 1, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.l_l3 = new piece(
            [0, 0, 0, 0, 0 ],
            [1, 1, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.l_l4 = new piece(
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 1, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.r_l1 = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 1, 0 ],
            [0, 1, 1, 1, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.r_l2 = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 1, 1, 0 ],
            [0, 0, 0, 1, 0 ],
            [0, 0, 0, 1, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.r_l3 = new piece(
            [0, 0, 0, 0, 0 ],
            [0, 0, 1, 1, 1 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
        piece.r_l4 = new piece(
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 0, 0 ],
            [0, 0, 1, 1, 0 ],
            [0, 0, 0, 0, 0 ],
            [0, 0, 0, 0, 0 ]        
        );
    
        
        var _board;
    
        var _interval;              // current frame interval
        var _speedinterval = 15000; // speed up every 15 sec

        var _intervalHandle;
    
        /*
            * This advances one frame
            */
        function tick() {
            _board.paint();
            if ( _board.autoUpdate() ) {
                _intervalHandle = setTimeout( tick, _interval);
            }
        }
    
        /*
            * This fires every 15 seconds and decreases the timeout of the other timer
            */
        function speedup() {
            if ( _interval > 20 )
                _interval -= 10;
        }
    
        function init(canvas) {
            _interval      = 200;

            _board = new board();
            _board.obtainContext(canvas);
        
            _intervalHandle = setTimeout( tick, _interval ); 
        }

        (function main(){
                        
            window.addEventListener('load', function() {
                var canvas = document.getElementById('tetris');

                canvas.addEventListener('click', function() {
                    if ( _intervalHandle )
                        clearTimeout( _intervalHandle );
                    init(canvas);
                });
                
                setInterval( speedup, _speedinterval );
            } );
    
            window.addEventListener( 
                'keydown', 
                function(e) {
                    //e.preventDefault();
                    if ( e.keyCode == 37 ) // arrow left
                        _board.moveCurrentPiece( pieceMove.left );
                    if ( e.keyCode == 39 ) // arrow right
                    _board.moveCurrentPiece( pieceMove.right );
                    if ( e.keyCode == 40 ) // arrow down
                        _board.moveCurrentPiece( pieceMove.down );
                    if ( e.keyCode == 32 ) // space
                        _board.moveCurrentPiece( pieceMove.rotate );
    
                    _board.paint();
                }
            );

        })();
    
    </script>
    <style>
        #welcome {
            border: 1px solid black;
            font: 10px tahoma;
            width: 100px;
        }
        #tetris {
            border: 1px solid black;
        }
    </style>    
</head>
<body>

    <div id="welcome">
        Moves: left/right/down/space
        Click: (re)start.
    </div>
    <div>
        <canvas id="tetris" width="100" height="400">
        </canvas>
    </div>

</body>
</html>

Topic Based Pub/Sub Data Bus with MassTransit

In my previous blog entry I've discussed the problem of using a single envelope-like data type in a data bus. I've presented a nice solution that uses topic-based subscriptions available in Rebus.FM framework. I've also promised to show how to do the same in MassTransit.

The problem with MassTransit is that you need a clever trick to set up all the bindings as MT doesn't support topic based subscriptions that easy. The trick is to have two layers of exchanges:

• the first layer of exchanges consists of a single exchange that is used by the publisher to put messages into the RabbitMQ
• the second layer of exchanges consists of multiple exchanges bound to the single exchange from the first layer by routing keys

and queues bound to exchanges from the second layer:

To get this, we need models

    public class MessageEnvelope
    {
        // will help to build routing keys
        public string MessageTypeName { get; set; }

        // the payload serialized to JSON
        public string Content { get; set; }

        public MessageEnvelope()
        {

        }

        public MessageEnvelope(object payload)
        {
            if (payload == null) throw new ArgumentNullException();

            this.Content         = JsonConvert.SerializeObject(payload);
            this.MessageTypeName = payload.GetType().Name;
        }

        public T GetPayload()
        {
            if (string.IsNullOrEmpty(this.Content)) throw new ArgumentException();

            return JsonConvert.DeserializeObject(this.Content);
        }
    }

    public static class EnvelopedExtensions
    {
        public static string ToJson(this object payload)
        {
            if (payload == null) throw new ArgumentNullException();

            return JsonConvert.SerializeObject(payload);
        }
    }

    public class Foo
    {
        public string Data { get; set; }
    }

    public class Bar
    {
        public string Something { get; set; }
    }

as well as the handler. Note that the handler is almost identical to the one from the previous blog entry about Rebus.FM. In fact, the only notable difference is naming of the interface you implement.

    public class Enveloped_Handler : IConsumer
    {
        private string agent { get; set; }
        public Enveloped_Handler(string agent)
        {
            this.agent = agent;
        }

        public async Task Consume(ConsumeContext context)
        {
            var message = context.Message;

            await Console.Out.WriteLineAsync($"{message.MessageTypeName} received by {this.agent}\r\n{message.ToJson()}");
        }
    }

Then, the actual code that setups the exchanges, queues and bindings and publishes messages.

   class Program
    {
        static IBusControl ConfigureActivator(string queueName, int workers = 1)
        {
            var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                var host = cfg.Host(new Uri("rabbitmq://localhost"), h =>
                {
                    h.Username("username");
                    h.Password("password");
                });

                if (queueName == "publisher")
                {
                    // take routing keys from the property
                    cfg.Send(x => x.UseRoutingKeyFormatter(context => context.Message.MessageTypeName ));
                    // publish to this exchange
                    cfg.Message(x => x.SetEntityName("MessageEnvelope"));
                    // use direct binding between the exchanges
                    cfg.Publish(x => { x.ExchangeType = ExchangeType.Direct; });
                }
                else
                {
                    cfg.ReceiveEndpoint(host, queueName, e =>
                    {
                        e.PrefetchCount = (ushort)workers;

                        switch (queueName)
                        {
                            case "subscriber1":

                                e.BindMessageExchanges = false;
                                e.Bind("MessageEnvelope", x =>
                                {
                                    x.ExchangeType = ExchangeType.Direct;
                                    x.RoutingKey = "Foo";
                                });


                                e.Consumer(() => new Enveloped_Handler("agent1"));
                                break;
                            case "subscriber2":

                                e.BindMessageExchanges = false;
                                e.Bind("MessageEnvelope", x =>
                                {
                                    x.ExchangeType = ExchangeType.Direct;
                                    x.RoutingKey = "Foo";
                                });
                                e.Bind("MessageEnvelope", x =>
                                {
                                    x.ExchangeType = ExchangeType.Direct;
                                    x.RoutingKey = "Bar";
                                });


                                e.Consumer(() => new Enveloped_Handler("agent2"));
                                break;
                        }
                        //e.con
                    });
                }
            });

            bus.Start();

            return bus;
        }

        static async Task Work()
        {
            var publisherBus   = ConfigureActivator("publisher");
            var subscriber1Bus = ConfigureActivator("subscriber1", 1);
            var subscriber2Bus = ConfigureActivator("subscriber2", 1);

            Console.WriteLine("publishing");

            await publisherBus.Publish(new MessageEnvelope( CreateFoo()) );
            await publisherBus.Publish(new MessageEnvelope( CreateBar()) );

            Console.WriteLine("published");
          
            Console.WriteLine("Press enter to quit");
            Console.ReadLine();
        }

        static Foo CreateFoo()
        {
            return new Foo()
            {
                Data = "hello foo world"
            };
        }

        static Bar CreateBar()
        {
            return new Bar()
            {
                Something = "and this is bar"
            };
        }

        static void Main(string[] args)
        {
            Work().Wait();
        } 
    }

The output of this code will be

publishing
published
Press enter to quit
Foo received by agent2
{"MessageTypeName":"Foo","Content":"{\"Data\":\"hello foo world\"}"}
Foo received by agent1
{"MessageTypeName":"Foo","Content":"{\"Data\":\"hello foo world\"}"}
Bar received by agent2
{"MessageTypeName":"Bar","Content":"{\"Something\":\"and this is bar\"}"}

If there's a simpler way to have this effect in MassTransit, in particular, if it's possible to have topic based subscriptions between exchanges and queues, with no this extra layer of exchanges, please let me know in the comments section.

Topic Based Pub/Sub Data Bus with Rebus.FM

We've been using messaging middlewares for years and it was always like "we don't want to be coupled to any specific implementation, rather we'd like to abstract the idea so that we could upgrade the internals when necessary". I remember first experiments with NServiceBus back in 2009/2010 but due to various reasons we've picked MassTransit. Using such higher-level approach has various advantages, for us, one of the thing was that we were able to migrate all deployments from MSMQ to RabbitMQ, thus gaining better performance and reliability.

One of issues that quickly turned out to be important is how the high-level engine uses the queuing subsystem. At the lower lever, where RabbitMQ operates, you have exchanges bound to queues and binary messages passed down the pipeline. At the higher level you have strongly typed contracts (classes) and strongly typed message handlers. A high-level engine needs a clean way of mapping its high level concepts to the wire level. Mass Transit does it in a straightforward manner:

Note: When a consumer is connected to a receive endpoint, the combined set of message types consumed by all of the consumers connected to the same receive endpoint are *subscribed* to the queue. The subscription method varies by broker, in the case of RabbitMQ exchange bindings are created for the message types to the exchange/queue for the receive endpoint. These subscriptions are persistent, and remain in place after the process exits. This ensures that messages published or sent that would be delivered to one of the receive endpoint consumers are saved even is the process is terminated. When the process is started, messages waiting in the queue will be delivered to the consumer(s).

What this actually means? Well, an exchange is created for each data type. In the diagram above we have two data types, Foo and Bar and thus two exchanges in RabbitMQ. Then, for each consumer, a queue is created. In the diagram we have two consumers and thus two queues.

When it comes to publishing, MassTransit just puts the message in the appropriate exchange. When it comes to subscription configuration, it uses the fanout model between exchanges and queues. In the above diagram, the Consumer 1 subscribes both message types while the Consumer 2 subscribes the Bar only. Consumers just read messages from their respective queues.

The model works great however it has its disadvantages. One of the major drawbacks here is how you set up subscriptions for consumers. It turns out that you write strongly typed handlers where the type has to be known in advance:

    public class Foo_Handler : IConsumer
    {
        public async Task Consume(ConsumeContext context)
        {
            var message = context.Message;

            // do whatever I like with and instance of Foo
        }
    }

This looks great, the handler can access the actual message in a strongly-typed fashion but ...

... all these handlers have to be either written down in an explicit way for each data type or I need a clever dynamic code generator to handle multiple data types. What's worse, my "bus" doesn't even sometimes know all the data types that are used, or rather - I don't want to recompile the data bus each time a new type appears in the family.

One of the possible approaches to the "family of data types that are not necessarily known and can change" issue involves a single "envelope-like" data type. A type that envelopes actual messages. Something like:

    // A generic envelope type
    public class MessageEnvelope
    {
        public string MessageTypeName { get; set; }

        // the actual payload serialized to JSON
        public string Content { get; set; }

        public MessageEnvelope()
        {

        }

        public MessageEnvelope(object payload)
        {
            if (payload == null) throw new ArgumentNullException();

            this.Content         = JsonConvert.SerializeObject(payload);
            this.MessageTypeName = payload.GetType().Name;
        }

        public T GetPayload()
        {
            if (string.IsNullOrEmpty(this.Content)) throw new ArgumentException();

            return JsonConvert.DeserializeObject(this.Content);
        }
    }

Suddenly, it's much easier to maintain the growing list of possible data types. I could handle types that are not known at the time the core library is compiled - the envelope will happily handle a payload of any type.

Unfortunately, while tempting, this approach has a significant disadvantage - please review the diagram above once again and imagine what happens in the RabbitMQ:

It's no longer that attractive. Now I end up with a single exchange that pushes incoming messages to all queues where agents pick up messages and have to filter them, according to the actual payload. And guess what, this slows down the pipeline! Suppose I have a lot of consumers, not two but, say, 50. A message of type Foo is published into the pipeline but since it's wrapped and is of the MessageEnvelope type, it's effectively copied to all queues (this is how the fanout works). Then, multiple consumers pick it just to realize that only few of them, maybe even a single one, subscribe this particular message. All other consumers just discard the message. Now, let's count. A single message, let's say few megabytes, copied by RabbitMQ to 50 different queues. Things go into hundred of megabytes and this is just a single message. Publish multiple messages and I have my bottleneck.

Fortunately, a solution exists that still lets me have a single envelope for all my messages but will not clog RabbitMQ with multiple messages. The solution involves topic mapping. I could still have a single exchange and multiple consumer queues. This time however, each message is published together with the topic and the binding between the exchange and queues involves explicit topis

How to set up the topic mapping in a high-level engine? Well, it's not that easy in MassTransit, unfortunately, and involves a trick I will present in another blog entry. This time I am going to show the actual code of how this particular set up works in Rebus, an engine much younger that the two introduced at the very beginning of this article. Younger doesn't mean it lacks features, though. What's nice is that it supports more transports than MassTransit (e.g. you don't even need a messaging middleware since the bare SQL Server can be used!) and supports topic based subscriptions easy.

Here's the code. We start with two dummy message types and the MessageEnvelope class presented above. A simple extension class will be used to format messages to the console.

    public class Foo
    {
        public string Data { get; set; }
    }

    public class Bar
    {
        public string Something { get; set; }
    }

    public static class EnvelopedExtensions
    {
        public static string ToJson( this object payload )
        {
            if (payload == null) throw new ArgumentNullException();

            return JsonConvert.SerializeObject(payload);
        }
    }

Now, the handler, the high-level concept of a consumer code that processes actual messages

    public class MessageEnvelope_Handler : IHandleMessages
    {
        private IBus bus { get; set; }
        private string agent { get; set; }
     
        public MessageEnvelope_Handler(IBus bus, string agent)
        {
            this.bus   = bus;
            this.agent = agent;
        }

        public async Task Handle(MessageEnvelope message)
        {
            var context = MessageContext.Current;

            await Console.Out.WriteLineAsync($"{message.MessageTypeName} received by {this.agent}\r\n{message.ToJson()}");
            
        }
    }

Then, a little configuration helper method to remove code duplication:

        static IBus ConfigureActivator(
            BuiltinHandlerActivator activator, 
            string queueName, 
            int workers = 1, 
            int parallel = 1)
        {
            var bus = Configure.With(activator)
                .Transport(t => t.UseRabbitMq("amqp://username:password@localhost", queueName))
                .Logging(l => l.ColoredConsole(LogLevel.Warn))
                .Options(conf =>
                {
                    conf.SetNumberOfWorkers(workers);
                    conf.SetMaxParallelism(parallel);
                })
                .Start();

            return bus;
        }

The actual orchestration

       static async Task Work()
        {
            using (var publisher   = new BuiltinHandlerActivator())
            using (var subscriber1 = new BuiltinHandlerActivator())
            using (var subscriber2 = new BuiltinHandlerActivator())
            {
                var publisherBus   = ConfigureActivator(publisher, "publisher");
                var subscriber1Bus = ConfigureActivator(subscriber1, "subscriber1", 2, 2);
                var subscriber2Bus = ConfigureActivator(subscriber2, "subscriber2", 2, 2);

                subscriber1.Register(() => new MessageEnvelope_Handler(subscriber1Bus, "agent1"));
                subscriber2.Register(() => new MessageEnvelope_Handler(subscriber2Bus, "agent2"));

                await subscriber1Bus.Advanced.Topics.Subscribe("Foo");
                await subscriber2Bus.Advanced.Topics.Subscribe("Foo");
                await subscriber2Bus.Advanced.Topics.Subscribe("Bar");

                Console.WriteLine("publishing");

                await publisherBus.Advanced.Topics.Publish("Foo", new MessageEnvelope(CreateFoo()));
                await publisherBus.Advanced.Topics.Publish("Bar", new MessageEnvelope(CreateBar()));

                Console.WriteLine("published");

                Console.WriteLine("Press enter to quit");
                Console.ReadLine();

            }
        }

As you can see, the Rebus' topic subscription API is used and then explicit topics are used to publish messages. Because of that, the very same message type, MessageEnvelope can be reused and still, RabbitMQ configures the binding correctly:

And this is the output of this simple demo:

publishing
published
Press enter to quit
Bar received by agent2
{"MessageTypeName":"Bar","Content":"{\"Something\":\"and this is bar\"}"}
Foo received by agent2
{"MessageTypeName":"Foo","Content":"{\"Data\":\"hello foo world\"}"}
Foo received by agent1
{"MessageTypeName":"Foo","Content":"{\"Data\":\"hello foo world\"}"}

This works as expected. The message of type Foo was handler by both consumers while Bar was handled by the second consumer only. At the price of the topic-based binding (vs the fanout binding) I have a fine-grained control over what happens in RabbitMQ and still have my single envelope type to wrap actual payloads.