Class: Valkey::Pipeline

Inherits:
Object
  • Object
show all
Includes:
Commands
Defined in:
lib/valkey/pipeline.rb

Overview

TODO: Pipeline/Multi routing is to be implemented. See https://github.com/valkey-io/valkey-glide-ruby/issues/137

Constant Summary

Constants included from Commands::TransactionCommands

Commands::TransactionCommands::BOOLEAN_REQUEST_TYPES

Constants included from Commands::ClusterCommands

Commands::ClusterCommands::CLUSTER_FAILOVER_MODES

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Commands::HashCommands

#hdel, #hexists, #hexpire, #hexpireat, #hexpiretime, #hget, #hgetall, #hgetex, #hincrby, #hincrbyfloat, #hkeys, #hlen, #hmget, #hmset, #hpersist, #hpexpire, #hpexpireat, #hpexpiretime, #hpttl, #hrandfield, #hscan, #hscan_each, #hset, #hsetex, #hsetnx, #hstrlen, #httl, #hvals, #mapped_hmget, #mapped_hmset

Methods included from Commands::StreamCommands

#xack, #xadd, #xautoclaim, #xclaim, #xdel, #xgroup, #xgroup_create, #xgroup_createconsumer, #xgroup_delconsumer, #xgroup_destroy, #xgroup_setid, #xinfo, #xinfo_consumers, #xinfo_groups, #xinfo_stream, #xlen, #xpending, #xrange, #xread, #xreadgroup, #xrevrange, #xtrim

Methods included from Commands::TransactionCommands

#discard, #exec, #multi, #unwatch, #watch

Methods included from Commands::ClusterCommands

#asking, #cluster_addslots, #cluster_addslotsrange, #cluster_bumpepoch, #cluster_count_failure_reports, #cluster_countkeysinslot, #cluster_delslots, #cluster_delslotsrange, #cluster_failover, #cluster_flushslots, #cluster_forget, #cluster_getkeysinslot, #cluster_info, #cluster_keyslot, #cluster_links, #cluster_meet, #cluster_myid, #cluster_myshardid, #cluster_nodes, #cluster_replicas, #cluster_replicate, #cluster_reset, #cluster_saveconfig, #cluster_set_config_epoch, #cluster_setslot, #cluster_shards, #cluster_slaves, #cluster_slots, #readonly, #readwrite

Methods included from Commands::PubSubCommands

#psubscribe, #publish, #pubsub, #pubsub_channels, #pubsub_numpat, #pubsub_numsub, #pubsub_shardchannels, #pubsub_shardnumsub, #punsubscribe, #spublish, #ssubscribe, #subscribe, #sunsubscribe, #unsubscribe

Methods included from Commands::FunctionCommands

#fcall, #fcall_ro, #function

Methods included from Commands::ScriptingCommands

#eval, #eval_ro, #evalsha, #evalsha_ro, #invoke_script, #script, #script_debug, #script_exists, #script_flush, #script_kill, #script_load

Methods included from Commands::SetCommands

#sadd, #sadd?, #scard, #sdiff, #sdiffstore, #sinter, #sinterstore, #sismember, #smembers, #smismember, #smove, #spop, #srandmember, #srem, #srem?, #sscan, #sscan_each, #sunion, #sunionstore

Methods included from Commands::SortedSetCommands

#bzmpop, #bzpopmax, #bzpopmin, #zadd, #zcard, #zcount, #zdiff, #zdiffstore, #zincrby, #zinter, #zintercard, #zinterstore, #zlexcount, #zmpop, #zmscore, #zpopmax, #zpopmin, #zrandmember, #zrange, #zrangebylex, #zrangebyscore, #zrangestore, #zrank, #zrem, #zremrangebylex, #zremrangebyrank, #zremrangebyscore, #zrevrange, #zrevrangebylex, #zrevrangebyscore, #zrevrank, #zscan, #zscore, #zunion, #zunionstore

Methods included from Commands::HyperLogLogCommands

#pfadd, #pfcount, #pfmerge

Methods included from Commands::GeoCommands

#geoadd, #geodist, #geohash, #geopos, #geosearch, #geosearchstore

Methods included from Commands::ListCommands

#blmove, #blmpop, #blpop, #brpop, #brpoplpush, #lindex, #linsert, #llen, #lmove, #lmpop, #lpop, #lpush, #lpushx, #lrange, #lrem, #lset, #ltrim, #rpop, #rpoplpush, #rpush, #rpushx

Methods included from Commands::BitmapCommands

#bitcount, #bitfield, #bitfield_ro, #bitop, #bitpos, #getbit, #setbit

Methods included from Commands::GenericCommands

#_scan, #copy, #del, #dump, #exists, #exists?, #expire, #expireat, #expiretime, #keys, #migrate, #move, #object, #persist, #pexpire, #pexpireat, #pexpiretime, #pttl, #rename, #renamenx, #restore, #scan, #scan_each, #sort, #touch, #ttl, #type, #unlink, #wait, #waitaof

Methods included from Commands::ServerCommands

#acl, #acl_cat, #acl_deluser, #acl_dryrun, #acl_genpass, #acl_getuser, #acl_list, #acl_load, #acl_log, #acl_save, #acl_setuser, #acl_users, #acl_whoami, #command, #command_, #command_count, #command_docs, #command_get_keys, #command_get_keys_and_flags, #command_info, #command_list, #config, #failover, #latency_doctor, #latency_graph, #latency_histogram, #latency_history, #latency_latest, #latency_reset, #memory_doctor, #memory_malloc_stats, #memory_purge, #memory_stats, #memory_usage, #psync, #replconf, #replicaof, #restore_asking, #role, #slaveof, #slowlog, #swapdb, #sync

Methods included from Commands::ConnectionCommands

#auth, #client, #client_get_name, #client_info, #client_kill, #client_kill_simple, #client_list, #client_no_evict, #client_no_touch, #client_pause, #client_set_info, #client_set_name, #hello, #reset, #select

Methods included from Commands::StringCommands

#append, #decr, #decrby, #get, #getdel, #getex, #getrange, #getset, #incr, #incrby, #incrbyfloat, #lcs, #mapped_mget, #mapped_mset, #mapped_msetnx, #mget, #mset, #msetnx, #psetex, #set, #setex, #setnx, #setrange, #strlen

Constructor Details

#initialize ⇒ Pipeline

Returns a new instance of Pipeline.



11
12
13
14
15
16
17
 
# File 'lib/valkey/pipeline.rb', line 11

def initialize
  @commands = []
  @futures = []
  # Keep transactional state consistent with the main client so that
  # helpers like `multi`/`exec` can safely consult `@in_multi`.
  @in_multi = false
end

Instance Attribute Details

#commands ⇒ Object (readonly)

Returns the value of attribute commands.



9
10
11
 
# File 'lib/valkey/pipeline.rb', line 9

def commands
  @commands
end

#futures ⇒ Object (readonly)

Returns the value of attribute futures.



9
10
11
 
# File 'lib/valkey/pipeline.rb', line 9

def futures
  @futures
end

Instance Method Details

#abort_futures! ⇒ Object

resolve_futures! ran, so every still-unresolved future raises a clear FutureAborted instead of hanging on FutureNotReady forever.



50
51
52
 
# File 'lib/valkey/pipeline.rb', line 50

def abort_futures!
  @futures.each(&:_abort!)
end

#bgrewriteaof ⇒ Object

rubocop:disable Lint/UselessMethodDefinition server_commands



65
66
67
 
# File 'lib/valkey/pipeline.rb', line 65

def bgrewriteaof
  super
end

#bgsave ⇒ Object 



69
70
71
 
# File 'lib/valkey/pipeline.rb', line 69

def bgsave
  super
end

#call(*argv, **kwargs) ⇒ Object 



176
177
178
179
180
181
182
 
# File 'lib/valkey/pipeline.rb', line 176

def call(*argv, **kwargs)
  if kwargs.key?(:route)
    raise ArgumentError, "Not supported: :route is not supported for individual pipelined commands"
  end

  super
end

#call_v(argv) ⇒ Object 



184
185
186
 
# File 'lib/valkey/pipeline.rb', line 184

def call_v(argv)
  super
end

#client_id ⇒ Object 



163
164
165
 
# File 'lib/valkey/pipeline.rb', line 163

def client_id
  super
end

#client_unpause ⇒ Object 



167
168
169
 
# File 'lib/valkey/pipeline.rb', line 167

def client_unpause
  super
end

#config_get(*args) ⇒ Object 



73
74
75
 
# File 'lib/valkey/pipeline.rb', line 73

def config_get(*args)
  super
end

#config_resetstat ⇒ Object 



81
82
83
 
# File 'lib/valkey/pipeline.rb', line 81

def config_resetstat
  super
end

#config_rewrite ⇒ Object 



85
86
87
 
# File 'lib/valkey/pipeline.rb', line 85

def config_rewrite
  super
end

#config_set(*args) ⇒ Object 



77
78
79
 
# File 'lib/valkey/pipeline.rb', line 77

def config_set(*args)
  super
end

#dbsize ⇒ Object 



89
90
91
 
# File 'lib/valkey/pipeline.rb', line 89

def dbsize
  super
end

#echo(value) ⇒ Object 



159
160
161
 
# File 'lib/valkey/pipeline.rb', line 159

def echo(value)
  super
end

#flushall(options = nil) ⇒ Object 



93
94
95
 
# File 'lib/valkey/pipeline.rb', line 93

def flushall(options = nil)
  super
end

#flushdb(options = nil) ⇒ Object 



97
98
99
 
# File 'lib/valkey/pipeline.rb', line 97

def flushdb(options = nil)
  super
end

#function_delete(library_name) ⇒ Object

function_commands



122
123
124
 
# File 'lib/valkey/pipeline.rb', line 122

def function_delete(library_name)
  super
end

#function_dump ⇒ Object 



126
127
128
 
# File 'lib/valkey/pipeline.rb', line 126

def function_dump
  super
end

#function_flush(async: false, sync: false) ⇒ Object 



130
131
132
 
# File 'lib/valkey/pipeline.rb', line 130

def function_flush(async: false, sync: false)
  super
end

#function_kill ⇒ Object 



134
135
136
 
# File 'lib/valkey/pipeline.rb', line 134

def function_kill
  super
end

#function_list(library_name: nil, with_code: false) ⇒ Object 



138
139
140
 
# File 'lib/valkey/pipeline.rb', line 138

def function_list(library_name: nil, with_code: false)
  super
end

#function_load(function_code, replace: false) ⇒ Object 



142
143
144
 
# File 'lib/valkey/pipeline.rb', line 142

def function_load(function_code, replace: false)
  super
end

#function_restore(serialized_value, policy: nil) ⇒ Object 



146
147
148
 
# File 'lib/valkey/pipeline.rb', line 146

def function_restore(serialized_value, policy: nil)
  super
end

#function_stats ⇒ Object 



150
151
152
 
# File 'lib/valkey/pipeline.rb', line 150

def function_stats
  super
end

#info(cmd = nil) ⇒ Object 



101
102
103
 
# File 'lib/valkey/pipeline.rb', line 101

def info(cmd = nil)
  super
end

#lastsave ⇒ Object 



105
106
107
 
# File 'lib/valkey/pipeline.rb', line 105

def lastsave
  super
end

#lolwut(version = nil) ⇒ Object 



117
118
119
 
# File 'lib/valkey/pipeline.rb', line 117

def lolwut(version = nil)
  super
end

#ping(message = nil) ⇒ Object

connection_commands



155
156
157
 
# File 'lib/valkey/pipeline.rb', line 155

def ping(message = nil)
  super
end

#randomkey ⇒ Object

generic_commands



172
173
174
 
# File 'lib/valkey/pipeline.rb', line 172

def randomkey
  super
end

#resolve_futures!(results) ⇒ Object

once send_batch_commands' final results are available. Purely positional, mirroring send_batch_commands' own per-command block re-application - safe for both the real-batch and the sequential MULTI/EXEC/DISCARD fallback branch, since both produce the same shape/order of results.

results is nil when a watched key was modified, aborting the whole transaction server-side before any queued command actually ran - none of these futures were ever really resolved, so treat it the same as abort_futures! instead of raising NoMethodError on a nil index.



41
42
43
44
45
 
# File 'lib/valkey/pipeline.rb', line 41

def resolve_futures!(results)
  return abort_futures! if results.nil?

  @futures.each_with_index { |future, i| future._set(results[i]) }
end

#save ⇒ Object 



109
110
111
 
# File 'lib/valkey/pipeline.rb', line 109

def save
  super
end

#send_command(command_type, command_args = [], route: nil, &block) ⇒ Object

route: is accepted and ignored. It exists for signature parity with Valkey#send_command, since any command reaching a pipeline through #call passes it, but a batch is dispatched as one unit so a per-command route cannot be honored. Batch-level routing is tracked in issue #137.



23
24
25
26
27
28
 
# File 'lib/valkey/pipeline.rb', line 23

def send_command(command_type, command_args = [], route: nil, &block) # rubocop:disable Lint/UnusedMethodArgument
  @commands << [command_type, command_args, block]
  future = Future.new(command_type, command_args)
  @futures << future
  future
end

#time ⇒ Object 



113
114
115
 
# File 'lib/valkey/pipeline.rb', line 113

def time
  super
end