Class: Libcouchbase::Bucket

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/libcouchbase/bucket.rb

Constant Summary collapse

AddDefaults = 
{operation: :add}.freeze
ReplaceDefaults = 
{operation: :replace}.freeze
AppendDefaults = 
{operation: :append}.freeze
PrependDefaults = 
{operation: :prepend}.freeze
ViewDefaults = 
{
    on_error: :stop,
    stale: false
}
FtsDefaults = 
{
    include_docs: true,
    size: 10000, # Max result size
    from: 0,
    explain: false
}

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(**options) ⇒ Bucket

Returns a new instance of Bucket.



20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/libcouchbase/bucket.rb', line 20

def initialize(**options)
    @connection_options = options
    @connection = Connection.new(**options)
    connect

    # This obtains the connections reactor
    @reactor = reactor
    @quiet = true

    # clean up the connection once this object is garbage collected
    ObjectSpace.define_finalizer( self, self.class.finalize(@connection) )
end

Instance Attribute Details

#connection ⇒ Object (readonly)

Returns the value of attribute connection.



34
35
36
 
# File 'lib/libcouchbase/bucket.rb', line 34

def connection
  @connection
end

#quiet ⇒ Object

Returns the value of attribute quiet.



35
36
37
 
# File 'lib/libcouchbase/bucket.rb', line 35

def quiet
  @quiet
end

Class Method Details

.finalize(connection) ⇒ Object

Finalizer done right http://www.mikeperham.com/2010/02/24/the-trouble-with-ruby-finalizers/



13
14
15
16
17
18
 
# File 'lib/libcouchbase/bucket.rb', line 13

def self.finalize(connection)
    proc {
        connection.reactor.unref
        connection.destroy
    }
end

Instance Method Details

#add(key, value, async: false, **opts) ⇒ Libcouchbase::Result

Add the item to the database, but fail if the object exists already

Examples:

Store the key which will be expired in 2 seconds using relative TTL.

c.add("foo", "bar", expire_in: 2)

Store the key which will be expired in 2 seconds using absolute TTL.

c.add(:foo, :bar, expire_at: Time.now.to_i + 2)

Force JSON document format for value

c.add("foo", {"bar" => "baz"}, format: :document)

Set application specific flags (note that it will be OR-ed with format flags)

c.add("foo", "bar", flags: 0x1000)

Ensure that the key will be persisted at least on the one node

c.add("foo", "bar", persist_to: 1)

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • value (Object) —

    Value to be stored

  • options (Hash) —

    Options for operation.

Returns:

  • (Libcouchbase::Result) —

    this includes the CAS value of the object.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue



202
203
204
 
# File 'lib/libcouchbase/bucket.rb', line 202

def add(key, value, async: false, **opts)
    result @connection.store(key, value, **AddDefaults.merge(opts)), async
end

#append(key, value, async: false, **opts) ⇒ Libcouchbase::Result

Note:

This operation is kind of data-aware from server point of view. This mean that the server treats value as binary stream and just perform concatenation, therefore it won't work with :marshal and :document formats, because of lack of knowledge how to merge values in these formats.

Append this object to the existing object

Examples:

Simple append

c.set(:foo, "aaa", format: :plain)
c.append(:foo, "bbb")
c.get("foo")           #=> "aaabbb"

Using optimistic locking. The operation will fail on CAS mismatch

resp = c.set("foo", "aaa", format: :plain)
c.append("foo", "bbb", cas: resp.cas)

Ensure that the key will be persisted at least on the one node

c.append("foo", "bar", persist_to: 1)

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • value (Object) —

    Value to be appended

  • options (Hash) —

    Options for operation.

Returns:

  • (Libcouchbase::Result) —

    this includes the CAS value of the object.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists



347
348
349
 
# File 'lib/libcouchbase/bucket.rb', line 347

def append(key, value, async: false, **opts)
    result @connection.store(key, value, **AppendDefaults.merge(opts)), async
end

#compare_and_swap(key, **opts) {|value| ... } ⇒ Libcouchbase::Response Also known as: cas

Compare and swap value.

Reads a key's value from the server and yields it to a block. Replaces the key's value with the result of the block as long as the key hasn't been updated in the meantime, otherwise raises Error::KeyExists.

Setting the :retry option to a positive number will cause this method to rescue the Error::KeyExists error that happens when an update collision is detected, and automatically get a fresh copy of the value and retry the block. This will repeat as long as there continues to be conflicts, up to the maximum number of retries specified.

Examples:

Implement append to JSON encoded value


c.set(:foo, {bar: 1})
c.cas(:foo) do |val|
  val[:baz] = 2
  val
end
c.get(:foo)      #=> {bar: 1, baz: 2}

Parameters:

  • key (String, Symbol)
  • options (Hash) —

    the options for "swap" part

Yield Parameters:

  • value (Object) —

    existing value

Yield Returns:

  • (Object) —

    new value.

Returns:

Raises:

  • (Couchbase::Error::KeyExists) —

    if the key was updated before the the code in block has been completed (the CAS value has been changed).

  • (ArgumentError) —

    if the block is missing



676
677
678
679
680
681
682
683
684
685
686
687
688
689
 
# File 'lib/libcouchbase/bucket.rb', line 676

def compare_and_swap(key, **opts)
    retries = opts.delete(:retry) || 0
    begin
        current = result(@connection.get(key))
        new_value = yield current.value, opts
        opts[:cas] = current.cas

        set(key, new_value, **opts)
    rescue Libcouchbase::Error::KeyExists
        retries -= 1
        retry if retries >= 0
        raise
    end
end

#decr(key, by = 1, **opts) ⇒ Object

Decrement the value of an existing numeric key

Helper method, see incr



458
459
460
 
# File 'lib/libcouchbase/bucket.rb', line 458

def decr(key, by = 1, **opts)
    incr(key, -by, **opts)
end

#delete(key, async: false, quiet: @quiet, **opts) ⇒ true, false

Delete the specified key

Examples:

Delete the key in quiet mode (default)

c.set("foo", "bar")
c.delete("foo")        #=> true
c.delete("foo")        #=> false

Delete the key verbosely

c.set("foo", "bar")
c.delete("foo", quiet: false)   #=> true
c.delete("foo", quiet: true)    #=> nil (default behaviour)
c.delete("foo", quiet: false)   #=> will raise Libcouchbase::Error::KeyNotFound

Delete the key with version check

res = c.set("foo", "bar")       #=> #<struct Libcouchbase::Response callback=:callback_set, key="foo", cas=1975457268957184, value="bar", metadata={:format=>:document, :flags=>0}>
c.delete("foo", cas: 123456)    #=> will raise Libcouchbase::Error::KeyExists
c.delete("foo", cas: res.cas)   #=> true

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • options (Hash) —

    Options for operation.

Returns:

  • (true, false) —

    the result of the operation.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists



498
499
500
501
502
503
504
505
506
507
508
509
510
 
# File 'lib/libcouchbase/bucket.rb', line 498

def delete(key, async: false, quiet: @quiet, **opts)
    promise = @connection.remove(key, **opts).then { true }
    if quiet
        promise = promise.catch { |error|
            if error.is_a? Libcouchbase::Error::KeyNotFound
                false
            else
                ::Libuv::Q.reject(@reactor, error)
            end
        }
    end
    result promise, async
end

#delete_design_doc(id, rev = nil, async: false) ⇒ Object

Delete design doc with given id and optional revision.

Parameters:

  • id (String, Symbol) —

    ID of the design doc

  • rev (String) (defaults to: nil) —

    Optional revision

See Also:



635
636
637
638
639
 
# File 'lib/libcouchbase/bucket.rb', line 635

def delete_design_doc(id, rev = nil, async: false)
    id = id.to_s.sub(/^_design\//, '')
    rev = "?rev=#{rev}" if rev
    result @connection.http("/_design/#{id}#{rev}", method: :delete, type: :view), async
end

#design_docs(**opts) ⇒ Libcouchbase::DesignDocs

Fetch design docs stored in current bucket

Returns:



535
536
537
 
# File 'lib/libcouchbase/bucket.rb', line 535

def design_docs(**opts)
    DesignDocs.new(self, @connection, method(:result), **opts)
end

#flush(async: false) ⇒ Libcouchbase::Response

Delete contents of the bucket

Examples:

Simple flush the bucket

c.flush

Returns:

Raises:

  • (Libcouchbase::Error::HttpError) —

    in case of an error is encountered.

See Also:



523
524
525
 
# File 'lib/libcouchbase/bucket.rb', line 523

def flush(async: false)
    result @connection.flush, async
end

#full_text_search(index, query, **opts, &row_modifier) ⇒ Libcouchbase::Results

Returns an enumerable for the results in a full text search.

Results are lazily loaded when an operation is performed on the enum

Returns:



569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
 
# File 'lib/libcouchbase/bucket.rb', line 569

def full_text_search(index, query, **opts, &row_modifier)
    if query.is_a? Hash
        opts[:query] = query
    else
        opts[:query] = {query: query}
    end
    fts = @connection.full_text_search(index, **FtsDefaults.merge(opts))

    current = ::Libuv::Reactor.current
    if current && current.running?
        ResultsLibuv.new(fts, current, &row_modifier)
    elsif Object.const_defined?(:EventMachine) && EM.reactor_thread?
        ResultsEM.new(fts, &row_modifier)
    else
        ResultsNative.new(fts, &row_modifier)
    end
end

#get(key, *keys, extended: false, async: false, quiet: @quiet, assemble_hash: false, **opts) ⇒ Object, ... Also known as: []

Obtain an object stored in Couchbase by given key.

Examples:

Get single value in quiet mode (the default)

c.get("foo")     #=> the associated value or nil

Use alternative hash-like syntax

c["foo"]         #=> the associated value or nil

Get single value in verbose mode

c.get("missing-foo", quiet: false)  #=> raises Libcouchbase::Error::NotFound

Get multiple keys

c.get("foo", "bar", "baz")   #=> [val1, val2, val3]

Get multiple keys with assembing result into the Hash

c.get("foo", "bar", "baz", assemble_hash: true)
#=> {"foo" => val1, "bar" => val2, "baz" => val3}

Get and lock key using default timeout

c.get("foo", lock: true)  # This locks for the maximum time of 30 seconds

Get and lock key using custom timeout

c.get("foo", lock: 3)

Get and lock multiple keys using custom timeout

c.get("foo", "bar", lock: 3)

Parameters:

  • keys (String, Symbol, Array) —

    One or several keys to fetch

  • options (Hash) —

    Options for operation.

Returns:

  • (Object, Array, Hash, Libcouchbase::Result) —

    the value(s)

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/libcouchbase/bucket.rb', line 87

def get(key, *keys, extended: false, async: false, quiet: @quiet, assemble_hash: false, **opts)
    was_array = key.respond_to?(:to_a) || keys.length > 0
    keys.unshift Array(key) # Convert enumerables
    keys.flatten!           # Ensure we're left with a list of keys

    if keys.length == 1
        promise = @connection.get(keys[0], **opts)

        unless extended
            promise = promise.then(proc { |resp|
                resp.value
            })
        end

        if quiet
            promise = promise.catch { |err|
                if err.is_a? Libcouchbase::Error::KeyNotFound
                    nil
                else
                    ::Libuv::Q.reject(@reactor, err)
                end
            }
        end

        if assemble_hash
            promise = promise.then(proc { |val|
                hash = defined?(::HashWithIndifferentAccess) ? ::HashWithIndifferentAccess.new : {}
                hash[keys[0]] = val
                hash
            })
        elsif was_array
            promise = promise.then(proc { |val|
                Array(val)
            })
        end

        result(promise, async)
    else
        promises = keys.collect { |key|
            @connection.get(key, **opts)
        }

        if quiet
            promises.map! { |prom|
                prom.catch { |err|
                    if err.is_a? Libcouchbase::Error::KeyNotFound
                        nil
                    else
                        ::Libuv::Q.reject(@reactor, err)
                    end
                }
            }
        end

        result(@reactor.all(*promises).then(proc { |results|
            if extended
                results.compact!
            else
                results.collect! { |resp| resp.value if resp }
            end

            if assemble_hash
                hash = defined?(::HashWithIndifferentAccess) ? ::HashWithIndifferentAccess.new : {}
                keys.each_with_index do |key, index|
                    hash[key] = results[index]
                end
                hash
            else
                results
            end
        }), async)
    end
end

#get_num_nodes ⇒ Integer

The numbers of nodes in the cluster

Returns:

  • (Integer) 


700
701
702
 
# File 'lib/libcouchbase/bucket.rb', line 700

def get_num_nodes
    result @connection.get_num_nodes
end

#get_num_replicas ⇒ Integer

The numbers of the replicas for each node in the cluster

Returns:

  • (Integer) 


694
695
696
 
# File 'lib/libcouchbase/bucket.rb', line 694

def get_num_replicas
    result @connection.get_num_replicas
end

#incr(key, by = 1, create: false, extended: false, async: false, **opts) ⇒ Integer

Increment the value of an existing numeric key

The increment method allow you to increase or decrease a given stored integer value. Updating the value of a key if it can be parsed to an integer. The update operation occurs on the server and is provided at the protocol level. This simplifies what would otherwise be a two-stage get and set operation.

Examples:

Increment key by one

c.incr(:foo)

Increment key by 50

c.incr("foo", 50)

Increment key by one OR initialize with zero

c.incr("foo", create: true)   #=> will return old+1 or 0

Increment key by one OR initialize with three

c.incr("foo", 50, initial: 3) #=> will return old+50 or 3

Increment key and get its CAS value

resp = c.incr("foo", :extended => true)
resp.cas   #=> 12345
resp.value #=> 2

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • by (Integer) (defaults to: 1) —

    Integer (up to 64 bits) value to increment or decrement

  • options (Hash) —

    Options for operation.

Returns:

  • (Integer) —

    the actual value of the key.

Raises:

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists

  • (Libcouchbase::Error::DeltaBadval) —

    if the key contains non-numeric value



445
446
447
448
449
450
451
452
453
 
# File 'lib/libcouchbase/bucket.rb', line 445

def incr(key, by = 1, create: false, extended: false, async: false, **opts)
    opts[:delta] ||= by
    opts[:initial] = 0 if create
    promise = @connection.counter(key, **opts)
    if not extended
        promise = promise.then { |resp| resp.value }
    end
    result promise, async
end

#n1ql(**options) ⇒ Libcouchbase::N1QL

Returns an n1ql query builder.

Returns:



596
597
598
 
# File 'lib/libcouchbase/bucket.rb', line 596

def n1ql(**options)
    N1QL.new(self, **options)
end

#prepend(key, value, async: false, **opts) ⇒ Libcouchbase::Result

Note:

This operation is kind of data-aware from server point of view. This mean that the server treats value as binary stream and just perform concatenation, therefore it won't work with :marshal and :document formats, because of lack of knowledge how to merge values in these formats.

Prepend this object to the existing object

Examples:

Simple prepend

c.set(:foo, "aaa", format: :plain)
c.prepend(:foo, "bbb")
c.get("foo")           #=> "bbbaaa"

Using optimistic locking. The operation will fail on CAS mismatch

resp = c.set("foo", "aaa", format: :plain)
c.prepend("foo", "bbb", cas: resp.cas)

Ensure that the key will be persisted at least on the one node

c.prepend("foo", "bar", persist_to: 1)

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • value (Object) —

    Value to be appended

  • options (Hash) —

    Options for operation.

Returns:

  • (Libcouchbase::Result) —

    this includes the CAS value of the object.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists



392
393
394
 
# File 'lib/libcouchbase/bucket.rb', line 392

def prepend(key, value, async: false, **opts)
    result @connection.store(key, value, **PrependDefaults.merge(opts)), async
end

#replace(key, value, async: false, **opts) ⇒ Libcouchbase::Result

Replace the existing object in the database

Examples:

Store the key which will be expired in 2 seconds using relative TTL.

c.replace("foo", "bar", expire_in: 2)

Store the key which will be expired in 2 seconds using absolute TTL.

c.replace(:foo, :bar, expire_at: Time.now.to_i + 2)

Force JSON document format for value

c.replace("foo", {"bar" => "baz"}, format: :document)

Set application specific flags (note that it will be OR-ed with format flags)

c.replace("foo", "bar", flags: 0x1000)

Ensure that the key will be persisted at least on the one node

c.replace("foo", "bar", persist_to: 1)

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • value (Object) —

    Value to be stored

  • options (Hash) —

    Options for operation.

Returns:

  • (Libcouchbase::Result) —

    this includes the CAS value of the object.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue

  • (Libcouchbase::Error::KeyNotFound) —

    if the key doesn't exists



302
303
304
 
# File 'lib/libcouchbase/bucket.rb', line 302

def replace(key, value, async: false, **opts)
    result @connection.store(key, value, **ReplaceDefaults.merge(opts)), async
end

#save_design_doc(data, id = nil, async: false) ⇒ Object

Update or create design doc with supplied views

Parameters:

  • data (Hash, IO, String) —

    The source object containing JSON encoded design document.

See Also:



606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
 
# File 'lib/libcouchbase/bucket.rb', line 606

def save_design_doc(data, id = nil, async: false)
    attrs = case data
    when String
        JSON.parse(data, Connection::DECODE_OPTIONS)
    when IO
        JSON.parse(data.read, Connection::DECODE_OPTIONS)
    when Hash
        data
    else
        raise ArgumentError, "Document should be Hash, String or IO instance"
    end
    attrs[:language] ||= :javascript

    id ||= attrs.delete(:_id)
    id = id.to_s.sub(/^_design\//, '')

    result @connection.http("/_design/#{id}",
        method: :put,
        body: attrs,
        type: :view
    ), async
end

#set(key, value, async: false, **opts) ⇒ Libcouchbase::Result Also known as: []=

Unconditionally store the object in the Couchbase

Examples:

Store the key which will be expired in 2 seconds using relative TTL.

c.set("foo", "bar", expire_in: 2)

Store the key which will be expired in 2 seconds using absolute TTL.

c.set(:foo, :bar, expire_at: Time.now.to_i + 2)

Force JSON document format for value

c.set("foo", {"bar" => "baz"}, format: :document)

Use hash-like syntax to store the value

c[:foo] = {bar: :baz}

Set application specific flags (note that it will be OR-ed with format flags)

c.set("foo", "bar", flags: 0x1000)

Perform optimistic locking by specifying last known CAS version

c.set("foo", "bar", cas: 8835713818674332672)

Ensure that the key will be persisted at least on the one node

c.set("foo", "bar", persist_to: 1)

Parameters:

  • key (String, Symbol) —

    Key used to reference the value.

  • value (Object) —

    Value to be stored

  • options (Hash) —

    Options for operation.

Returns:

  • (Libcouchbase::Result) —

    this includes the CAS value of the object.

Raises:

  • (Libcouchbase::Error::KeyExists) —

    if the key already exists on the server with a different CAS value to that provided

  • (Libouchbase::Error::Timedout) —

    if timeout interval for observe exceeds

  • (Libouchbase::Error::NetworkError) —

    if there was a communication issue



254
255
256
257
 
# File 'lib/libcouchbase/bucket.rb', line 254

def set(key, value, async: false, **opts)
    # default operation is set
    result @connection.store(key, value, **opts), async
end

#touch(async: false, **opts) ⇒ Object

Touch a key, changing its CAS and optionally setting a timeout



528
529
530
 
# File 'lib/libcouchbase/bucket.rb', line 528

def touch(async: false, **opts)
    result @connection.touch(**opts), async
end

#view(design, view, include_docs: true, is_spatial: false, **opts, &row_modifier) ⇒ Libcouchbase::Results

Returns an enumerable for the results in a view.

Results are lazily loaded when an operation is performed on the enum

Returns:



544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
 
# File 'lib/libcouchbase/bucket.rb', line 544

def view(design, view, include_docs: true, is_spatial: false, **opts, &row_modifier)
    view = @connection.query_view(design, view, **ViewDefaults.merge(opts))
    view.include_docs = include_docs
    view.is_spatial = is_spatial

    current = ::Libuv::Reactor.current

    if current && current.running?
        ResultsLibuv.new(view, current, &row_modifier)
    elsif Object.const_defined?(:EventMachine) && EM.reactor_thread?
        ResultsEM.new(view, &row_modifier)
    else
        ResultsNative.new(view, &row_modifier)
    end
end

#wait_results(*results) ⇒ Array

Waits for all the async operations to complete and returns the results

Returns:

  • (Array) 


707
708
709
 
# File 'lib/libcouchbase/bucket.rb', line 707

def wait_results(*results)
    result ::Libuv::Q.all(@reactor, *results.flatten)
end