bucket.rb

# Author:: Couchbase <info@couchbase.com>
# Copyright:: 2011, 2012 Couchbase, Inc.
# License:: Apache License, Version 2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

module Couchbase

  class Bucket

    # Compare and swap value.
    #
    # @since 1.0.0
    #
    # 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
    # {Couchbase::Error::KeyExists}. CAS stands for "compare and swap", and
    # avoids the need for manual key mutexing. Read more info here:
    #
    # In asynchronous mode it will yield result twice, first for
    # {Bucket#get} with {Result#operation} equal to +:get+ and
    # second time for {Bucket#set} with {Result#operation} equal to +:set+.
    #
    # @see http://couchbase.com/docs/memcached-api/memcached-api-protocol-text_cas.html
    #
    # @param [String, Symbol] key
    #
    # @param [Hash] options the options for "swap" part
    # @option options [Fixnum] :ttl (self.default_ttl) the time to live of this key
    # @option options [Symbol] :format (self.default_format) format of the value
    # @option options [Fixnum] :flags (self.default_flags) flags for this key
    #
    # @yieldparam [Object, Result] value old value in synchronous mode and
    #   +Result+ object in asynchronous mode.
    # @yieldreturn [Object] new value.
    #
    # @raise [Couchbase::Error::KeyExists] if the key was updated before the the
    #   code in block has been completed (the CAS value has been changed).
    # @raise [ArgumentError] if the block is missing for async mode
    #
    # @example Implement append to JSON encoded value
    #
    #     c.default_format = :document
    #     c.set("foo", {"bar" => 1})
    #     c.cas("foo") do |val|
    #       val["baz"] = 2
    #       val
    #     end
    #     c.get("foo")      #=> {"bar" => 1, "baz" => 2}
    #
    # @example Append JSON encoded value asynchronously
    #
    #     c.default_format = :document
    #     c.set("foo", {"bar" => 1})
    #     c.run do
    #       c.cas("foo") do |val|
    #         case val.operation
    #         when :get
    #           val["baz"] = 2
    #           val
    #         when :set
    #           # verify all is ok
    #           puts "error: #{ret.error.inspect}" unless ret.success?
    #         end
    #       end
    #     end
    #     c.get("foo")      #=> {"bar" => 1, "baz" => 2}
    #
    # @return [Fixnum] the CAS of new value
    def cas(key, options = {})
      if async?
        block = Proc.new
        get(key) do |ret|
          val = block.call(ret) # get new value from caller
          set(ret.key, val, options.merge(:cas => ret.cas, :flags => ret.flags), &block)
        end
      else
        val, flags, ver = get(key, :extended => true)
        val = yield(val) # get new value from caller
        set(key, val, options.merge(:cas => ver, :flags => flags))
      end
    end
    alias :compare_and_swap :cas

    # Fetch design docs stored in current bucket
    #
    # @since 1.2.0
    #
    # @return [Hash]
    def design_docs
      docs = all_docs(:startkey => "_design/", :endkey => "_design0", :include_docs => true)
      docmap = {}
      docs.each do |doc|
        key = doc.id.sub(/^_design\//, '')
        next if self.environment == :production && key =~ /dev_/
        docmap[key] = doc
      end
      docmap
    end

    # Fetch all documents from the bucket.
    #
    # @since 1.2.0
    #
    # @param [Hash] params Params for Couchbase +/_all_docs+ query
    #
    # @return [Couchbase::View] View object
    def all_docs(params = {})
      View.new(self, "_all_docs", params)
    end

    # Update or create design doc with supplied views
    #
    # @since 1.2.0
    #
    # @param [Hash, IO, String] data The source object containing JSON
    #   encoded design document. It must have +_id+ key set, this key
    #   should start with +_design/+.
    #
    # @return [true, false]
    def save_design_doc(data)
      attrs = case data
              when String
                MultiJson.load(data)
              when IO
                MultiJson.load(data.read)
              when Hash
                data
              else
                raise ArgumentError, "Document should be Hash, String or IO instance"
              end

      if attrs['_id'].to_s !~ /^_design\//
        raise ArgumentError, "'_id' key must be set and start with '_design/'."
      end
      attrs['language'] ||= 'javascript'
      req = make_couch_request(attrs['_id'],
                               :body => MultiJson.dump(attrs),
                               :method => :put)
      res = MultiJson.load(req.perform)
      if res['ok']
        true
      else
        raise "Failed to save design document: #{res['error']}"
      end
    end

    # Delete design doc with given id and revision.
    #
    # @since 1.2.0
    #
    # @param [String] id Design document id. It might have '_design/'
    #   prefix.
    #
    # @param [String] rev Document revision. It uses latest revision if
    #   +rev+ parameter is nil.
    #
    # @return [true, false]
    def delete_design_doc(id, rev = nil)
      ddoc = design_docs[id.sub(/^_design\//, '')]
      return nil unless ddoc
      path = Utils.build_query(ddoc['_id'], :rev => rev || ddoc['_rev'])
      req = make_couch_request(path, :method => :delete)
      res = MultiJson.load(req.perform)
      if res['ok']
        true
      else
        raise "Failed to save design document: #{res['error']}"
      end
    end

    def create_timer(interval, &block)
      Timer.new(self, interval, &block)
    end

    def create_periodic_timer(interval, &block)
      Timer.new(self, interval, :periodic => true, &block)
    end

    # Wait for persistence condition
    #
    # @since 1.2.0.dp6
    #
    # This operation is useful when some confidence needed regarding the
    # state of the keys. With two parameters +:replicated+ and +:persisted+
    # it allows to set up the waiting rule.
    #
    # @param [String, Symbol, Array, Hash] keys The list of the keys to
    #   observe. Full form is hash with key-cas value pairs, but there are
    #   also shortcuts like just Array of keys or single key. CAS value
    #   needed to when you need to ensure that the storage persisted exactly
    #   the same version of the key you are asking to observe.
    # @param [Hash] options The options for operation
    # @option options [Fixnum] :timeout The timeout in microseconds
    # @option options [Fixnum] :replicated How many replicas should receive
    #   the copy of the key.
    # @option options [Fixnum] :persisted How many nodes should store the
    #   key on the disk.
    #
    # @return [Couchbase::Result, Array<Couchbase::Result>] will return
    #   single result object in case the keys argument is String or Symbol.
    def observe_and_wait(*keys, &block)
      options = {:timeout => default_observe_timeout}
      options.update(keys.pop) if keys.size > 1 && keys.last.is_a?(Hash)
      verify_observe_options(options)
      if block && !async?
        raise ArgumentError, "synchronous mode doesn't support callbacks"
      end
      if keys.size == 0
        raise ArgumentError, "at least one key is required"
      end
      if keys.size == 1 && keys[0].is_a?(Hash)
        key_cas = keys[0]
      else
        key_cas = keys.flatten.reduce({}) do |h, kk|
          h[kk] = nil   # set CAS to nil
          h
        end
      end
      if async?
        do_observe_and_wait(key_cas, options, &block)
      else
        res = do_observe_and_wait(key_cas, options, &block) while res.nil?
        if keys.size == 1 && (keys[0].is_a?(String) || keys[0].is_a?(Symbol))
          return res[0]
        else
          return res
        end
      end
    end

    private

    def verify_observe_options(options)
      unless num_replicas
        raise Couchbase::Error::Libcouchbase, "cannot detect number of the replicas"
      end
      unless options[:persisted] || options[:replicated]
        raise ArgumentError, "either :persisted or :replicated option must be set"
      end
      if options[:persisted] && !(1..num_replicas + 1).include?(options[:persisted])
        raise ArgumentError, "persisted number should be in range (1..#{num_replicas + 1})"
      end
      if options[:replicated] && !(1..num_replicas).include?(options[:replicated])
        raise ArgumentError, "replicated number should be in range (1..#{num_replicas})"
      end
    end

    def do_observe_and_wait(keys, options, &block)
      acc = Hash.new do |h, k|
        h[k] = Hash.new(0)
        h[k][:cas] = [keys[k]] # first position is for master node
        h[k]
      end
      check_condition = lambda do
        ok = catch :break do
          acc.each do |key, stats|
            master = stats[:cas][0]
            if master.nil?
              # master node doesn't have the key
              throw :break
            end
            if options[:persisted] && (stats[:persisted] < options[:persisted] ||
                                       stats[:cas].count(master) != options[:persisted])
              throw :break
            end
            if options[:replicated] && (stats[:replicated] < options[:replicated] ||
                                        stats[:cas].count(master) != options[:replicated] + 1)
              throw :break
            end
          end
          true
        end
        if ok
          res = keys.map do |k, _|
            Result.new(:key => k, :cas => acc[k][:cas][0], :operation => :observe_and_wait)
          end
          if async?
            options[:timer].cancel if options[:timer]
            block.call(res)
          else
            return res
          end
        else
          options[:timeout] /= 2
          if options[:timeout] > 0
            if async?
              options[:timer] = create_timer(options[:timeout]) do
                do_observe_and_wait(keys, options, &block)
              end
            else
              # do wait for timeout
              run { create_timer(options[:timeout]){} }
              # return nil to avoid recursive call
              return nil
            end
          else
            err = Couchbase::Error::Timeout.new("the observe request was timed out")
            err.instance_variable_set("@operation", :observe_and_wait)
            if async?
              res = keys.map do |k, _|
                Result.new(:key => k, :cas => acc[k][:cas][0], :operation => :observe_and_wait)
              end
              block.call(res)
            else
              err.instance_variable_set("@key", keys.keys)
              raise err
            end
          end
        end
      end
      collect = lambda do |res|
        if res.completed?
          check_condition.call if async?
        else
          if res.from_master?
            acc[res.key][:cas][0] = res.cas
          else
            acc[res.key][:cas] << res.cas
          end
          acc[res.key][res.status] += 1
          if res.status == :persisted
            acc[res.key][:replicated] += 1
          end
        end
      end
      if async?
        observe(*(keys.keys), options, &collect)
      else
        observe(*(keys.keys), options).each(&collect)
        check_condition.call
      end
    end
  end

end