hunt.redis.ShardedRedisPipeline source code

1 /*
2  * Hunt - A redis client library for D programming language.
3  *
4  * Copyright (C) 2018-2019 HuntLabs
5  *
6  * Website: https://www.huntlabs.net/
7  *
8  * Licensed under the Apache-2.0 License.
9  *
10  */
11  
12 module hunt.redis.ShardedRedisPipeline;
13 
14 import hunt.redis.Client;
15 import hunt.redis.BinaryShardedRedis;
16 import hunt.redis.PipelineBase;
17 
18 import hunt.collection;
19 import hunt.Exceptions;
20 
21 
22 private class FutureResult {
23     private Client client;
24 
25     this(Client client) {
26         this.client = client;
27     }
28 
29     Object get() {
30         return client.getOne();
31     }
32 }
33 
34 /**
35 */
36 class ShardedRedisPipeline : PipelineBase {
37     private BinaryShardedRedis jedis;
38     private List!(FutureResult) results;
39     private Queue!(Client) clients;
40 
41     this() {
42         results = new ArrayList!(FutureResult)();
43         clients = new LinkedList!(Client)();
44     }
45 
46 
47     void setShardedRedis(BinaryShardedRedis jedis) {
48         this.jedis = jedis;
49     }
50 
51     List!(Object) getResults() {
52         List!(Object) r = new ArrayList!(Object)();
53         foreach(FutureResult fr ; results) {
54             r.add(fr.get());
55         }
56         return r;
57     }
58 
59     /**
60      * Synchronize pipeline by reading all responses. This operation closes the pipeline. In order to
61      * get return values from pipelined commands, capture the different Response&lt;?&gt; of the
62      * commands you execute.
63      */
64     void sync() {
65         foreach(Client client ; clients) {
66             generateResponse(client.getOne());
67         }
68     }
69 
70     /**
71      * Synchronize pipeline by reading all responses. This operation closes the pipeline. Whenever
72      * possible try to avoid using this version and use ShardedRedisPipeline.sync() as it won't go
73      * through all the responses and generate the right response type (usually it is a waste of time).
74      * @return A list of all the responses in the order you executed them.
75      */
76     List!(Object) syncAndReturnAll() {
77         // List!(Object) formatted = new ArrayList!(Object)();
78         // foreach(Client client ; clients) {
79         //     formatted.add(generateResponse(client.getOne()).get());
80         // }
81         // return formatted;
82         implementationMissing(false);
83         return null;
84     }
85 
86     override
87     protected Client getClient(string key) {
88         Client client = jedis.getShard(key).getClient();
89         clients.add(client);
90         results.add(new FutureResult(client));
91         return client;
92     }
93 
94     override
95     protected Client getClient(const(ubyte)[] key) {
96         Client client = jedis.getShard(key).getClient();
97         clients.add(client);
98         results.add(new FutureResult(client));
99         return client;
100     }
101 }