lock/rpc.go at master · roadrunner-server/lock · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
package lock

import (
	"context"
	"errors"
	"time"

	lockV1 "github.com/roadrunner-server/api-go/v6/lock/v1"
)

const defaultImmediateTimeout = time.Millisecond

var errEmptyID = errors.New("empty ID is not allowed")

type rpc struct {
	pl *Plugin
}

func waitContext(parent context.Context, waitUs int64) (context.Context, context.CancelFunc) {
	if waitUs == 0 {
		return context.WithTimeout(parent, defaultImmediateTimeout)
	}
	return context.WithTimeout(parent, time.Microsecond*time.Duration(waitUs))
}

func (r *rpc) Lock(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("lock request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	if in.GetId() == "" {
		return errEmptyID
	}

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.lock(cctx, in.GetResource(), in.GetId(), int(in.GetTtl()))
	return nil
}

func (r *rpc) LockRead(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("read lock request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	if in.GetId() == "" {
		return errEmptyID
	}

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.lockRead(cctx, in.GetResource(), in.GetId(), int(in.GetTtl()))
	return nil
}

func (r *rpc) Release(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("release request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	if in.GetId() == "" {
		return errEmptyID
	}

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.release(cctx, in.GetResource(), in.GetId())
	return nil
}

func (r *rpc) ForceRelease(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("force release request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.forceRelease(cctx, in.GetResource())
	return nil
}

func (r *rpc) Exists(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("exists request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	if in.GetId() == "" {
		return errEmptyID
	}

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.exists(cctx, in.GetResource(), in.GetId())
	return nil
}

func (r *rpc) UpdateTTL(in *lockV1.LockRequest, out *lockV1.LockResponse) error {
	r.pl.log.Debug("updateTTL request received", "ttl", int(in.GetTtl()), "wait_ttl", int(in.GetWait()), "resource", in.GetResource(), "id", in.GetId())

	if in.GetId() == "" {
		return errEmptyID
	}

	cctx, cancel := waitContext(context.Background(), in.GetWait())
	defer cancel()

	out.Ok = r.pl.locks.updateTTL(cctx, in.GetResource(), in.GetId(), int(in.GetTtl()))
	return nil
}