elasticell
/
pkg
/
raftstore
/
redis_hash.go

// Copyright 2016 DeepFabric, Inc.
//
// 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,
// See the License for the specific language governing permissions and
// limitations under the License.

package raftstore

import (
	"gitee.com/wanttobeamaster/elasticell/pkg/pb/raftcmdpb"
	"gitee.com/wanttobeamaster/elasticell/pkg/pool"
	"gitee.com/wanttobeamaster/elasticell/pkg/redis"
	"github.com/fagongzi/util/format"
	"github.com/fagongzi/util/hack"
)

func (s *Store) execHSet(ctx *applyContext, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 3 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	n, err := s.getHashEngine(ctx.req.Header.CellId).HSet(args[0], args[1], args[2])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	if n > 0 {
		size := uint64(len(args[1]) + len(args[2]))
		ctx.metrics.writtenKeys++
		ctx.metrics.writtenBytes += size
		ctx.metrics.sizeDiffHint += size
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &n
	return rsp
}

func (s *Store) execHDel(ctx *applyContext, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) < 2 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	n, err := s.getHashEngine(ctx.req.Header.CellId).HDel(args[0], args[1:]...)
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	if n > 0 {
		var size uint64

		for _, arg := range args[1:] {
			size += uint64(len(arg))
		}

		ctx.metrics.sizeDiffHint -= size
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &n
	return rsp
}

func (s *Store) execHMSet(ctx *applyContext, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()
	var size uint64

	l := len(args)
	if l < 3 || l%2 == 0 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	key := args[0]
	kvs := args[1:]
	l = len(kvs) / 2
	fields := make([][]byte, l)
	values := make([][]byte, l)

	for i := 0; i < l; i++ {
		fields[i] = kvs[2*i]
		values[i] = kvs[2*i+1]

		size += uint64(len(fields[i]))
		size += uint64(len(values[i]))
	}

	err := s.getHashEngine(ctx.req.Header.CellId).HMSet(key, fields, values)
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	ctx.metrics.writtenKeys++
	ctx.metrics.writtenBytes += size
	ctx.metrics.sizeDiffHint += size

	rsp := pool.AcquireResponse()
	rsp.StatusResult = redis.OKStatusResp

	return rsp
}

func (s *Store) execHSetNX(ctx *applyContext, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 3 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(ctx.req.Header.CellId).HSetNX(args[0], args[1], args[2])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	if value > 0 {
		size := uint64(len(args[1]) + len(args[2]))
		ctx.metrics.writtenBytes += size
		ctx.metrics.sizeDiffHint += size
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &value
	return rsp
}

func (s *Store) execHIncrBy(ctx *applyContext, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 3 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	incrment, err := format.ParseStrInt64(hack.SliceToString(args[2]))
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	value, err := s.getHashEngine(ctx.req.Header.CellId).HIncrBy(args[0], args[1], incrment)
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	v, err := format.ParseStrInt64(hack.SliceToString(value))
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &v
	return rsp
}

func (s *Store) execHGet(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 2 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HGet(args[0], args[1])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.BulkResult = value
	rsp.HasEmptyBulkResult = len(value) == 0

	return rsp
}

func (s *Store) execHExists(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 2 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	exists, err := s.getHashEngine(id).HExists(args[0], args[1])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	var value int64
	if exists {
		value = 1
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &value
	return rsp
}

func (s *Store) execHKeys(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 1 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HKeys(args[0])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.SliceArrayResult = value
	rsp.HasEmptySliceArrayResult = len(value) == 0

	return rsp
}

func (s *Store) execHVals(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 1 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HVals(args[0])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.SliceArrayResult = value
	rsp.HasEmptySliceArrayResult = len(value) == 0

	return rsp
}

func (s *Store) execHGetAll(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 1 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HGetAll(args[0])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.FvPairArrayResult = value
	rsp.HasEmptyFVPairArrayResult = len(value) == 0

	return rsp
}

func (s *Store) execHScanGet(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 3 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp
		return rsp
	}

	count, err := format.ParseStrInt64(hack.SliceToString(args[2]))
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())
		return rsp
	}

	value, err := s.getHashEngine(id).HScanGet(args[0], args[1], int(count))
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())
		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.FvPairArrayResult = value
	rsp.HasEmptyFVPairArrayResult = len(value) == 0

	return rsp
}

func (s *Store) execHLen(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 1 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HLen(args[0])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &value
	return rsp
}

func (s *Store) execHMGet(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) < 2 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, errs := s.getHashEngine(id).HMGet(args[0], args[1:]...)
	if len(errs) > 0 {
		errors := make([][]byte, len(errs))
		for idx, err := range errs {
			errors[idx] = hack.StringToSlice(err.Error())
		}

		rsp := pool.AcquireResponse()
		rsp.ErrorResults = errors

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.SliceArrayResult = value
	rsp.HasEmptySliceArrayResult = len(value) == 0

	return rsp

}

func (s *Store) execHStrLen(id uint64, req *raftcmdpb.Request) *raftcmdpb.Response {
	cmd := redis.Command(req.Cmd)
	args := cmd.Args()

	if len(args) != 2 {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = redis.ErrInvalidCommandResp

		return rsp
	}

	value, err := s.getHashEngine(id).HStrLen(args[0], args[1])
	if err != nil {
		rsp := pool.AcquireResponse()
		rsp.ErrorResult = hack.StringToSlice(err.Error())

		return rsp
	}

	rsp := pool.AcquireResponse()
	rsp.IntegerResult = &value
	return rsp
}