Validate key slots used with RedisCluster::Client#with

[KJTsanaktsidis]

The overall objective of this PR is to unskip the test which asserts that TransactionConsistencyError is raised when using a key belonging to the wrong slot on a connection returned from #with. i.e. it makes this test pass:

        assert_raises(::RedisClient::Cluster::Transaction::ConsistencyError) do
          @client.with(hashtag: 'slot1') do |conn|
            conn.call('GET', '{slot2}')
          end
        end

In order to get there, though, there is a bit of a journey. I've broken it up into five separate commits, which I can make separate PR's for - I did want to show you the whole picture before we drill down on the detail however.

Firstly, and sadly, I discovered that we do actually need to make a custom subclass of RedisClient, and pass that as client_implementation: in our custom config class. I had previously thought I could implement the slot validation just with middleware, but we need a way to communicate between #with, which just has access to the client object, and the middleware. By implementing a custom subclass, we can define a method Client#locked_to_key_slot, which essentially turns on and off the middleware by setting a flag on the client object which the middleware can see.

I do realise that we'd discussed doing this without the client subclass, but unfortunately I could not find another way.

The next phase of this PR involves getting access to the RedisCluster::Client::Command object from inside the middleware. There's a bit of circularity here, because obviously we need clients to run COMMAND on, but we need the Command object inside the client middleware (to be able to call #extract_all_keys on passed in commands). I resolved this by turning Command into a long-lived object which gets passed into the Node::Config redisclient config objects.

Once all that's done, we're finally in a position to actually implement the "lock to a single slot"; that then turns out to be pretty simple in a single file pinning.rb; we just check if the connection is currently supposed to be pinned, and compare the pinned slot against Command#extract_all_keys(command) if so.

Lets debate the general direction of this PR, and I can cut it up into smaller pieces for more detailed review once we're OK with that perhaps?

[KJTsanaktsidis]

Poke @supercaracal wondering if you're able to have a look this week? 🙏 I anticipate this is going to need some feedback and reworking :)

[supercaracal]

Sorry for my delayed response. I'm going to review this pull request within a week.

[supercaracal]

One more thing, I think it would be better to be able to enable or disable the client side validation feature by options.

class RedisClient
  class Cluster
    def with(key: nil, hashtag: nil, write: true, retry_count: 0)
      key = process_with_arguments(key, hashtag)
      node_key = @router.find_node_key_by_key(key, primary: write)
      node = @router.find_node(node_key)
      @router.try_delegate(node, :with, retry_count: retry_count) do |conn|
        if @config.client_side_validation_enabled?
          yield PinningClient.new(conn, key, @command_builder, @router.command)
        else
          yield conn
        end
      end
    end
  end
end

[KJTsanaktsidis]

Thanks for your feedback @supercaracal - I opened https://github.com/redis-rb/redis-cluster-client/pull/317 instead which makes the changes you suggested here (it's so different I figured it was better to open a separate PR rather than clutter this one).