update to lcw@master

This commit is contained in:
Umputun
2020-05-09 21:20:03 -05:00
parent bbfeaeb3d0
commit a8f3e1730b
41 changed files with 1158 additions and 458 deletions
+1 -1
View File
@@ -14,7 +14,7 @@ require (
github.com/go-chi/render v1.0.1
github.com/go-pkgz/auth v0.10.1
github.com/go-pkgz/jrpc v0.1.0
github.com/go-pkgz/lcw v0.5.0
github.com/go-pkgz/lcw v0.5.1-0.20200509170726-dc283cfc28cf
github.com/go-pkgz/lgr v0.7.0
github.com/go-pkgz/repeater v1.1.3
github.com/go-pkgz/rest v1.5.0
+22 -20
View File
@@ -25,8 +25,8 @@ github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1 h1:GDQdwm/gAcJcLAK
github.com/alecthomas/repr v0.0.0-20181024024818-d37bc2a10ba1/go.mod h1:xTS7Pm1pD1mvyM075QCDSRqH6qRLXylzS24ZTpRiSzQ=
github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6 h1:45bxf7AZMwWcqkLzDAQugVEwedisr5nRJ1r+7LYnv0U=
github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
github.com/alicebob/miniredis v2.5.0+incompatible h1:yBHoLpsyjupjz3NL3MhKMVkR41j82Yjf3KFv7ApYzUI=
github.com/alicebob/miniredis v2.5.0+incompatible/go.mod h1:8HZjEj4yU0dwhYHky+DxYx+6BMjkBbe5ONFIF1MXffk=
github.com/alicebob/miniredis/v2 v2.11.4 h1:GsuyeunTx7EllZBU3/6Ji3dhMQZDpC9rLf1luJ+6M5M=
github.com/alicebob/miniredis/v2 v2.11.4/go.mod h1:VL3UDEfAH59bSa7MuHMuFToxkqyHh69s/WUbYlOAuyg=
github.com/andybalholm/cascadia v1.1.0 h1:BuuO6sSfQNFRu1LppgbD25Hr2vLYW25JvxHs5zzsLTo=
github.com/andybalholm/cascadia v1.1.0/go.mod h1:GsXiBklL0woXo1j/WYWtSYYC4ouU9PqHO0sqidkEA4Y=
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
@@ -68,8 +68,8 @@ github.com/go-pkgz/auth v0.10.1 h1:GYf64js5n/oVEPXhRShyDDiR/oahdZOMLbxmnnGX2IU=
github.com/go-pkgz/auth v0.10.1/go.mod h1:wxyQqc0UUP1jT4l6zk1r6XPcVdcgIzW2OiQ8hBEHd64=
github.com/go-pkgz/jrpc v0.1.0 h1:hNg/IyfEqJcSWOKkuHw0ZwcuGc9TDp7QZREsD2ycmiM=
github.com/go-pkgz/jrpc v0.1.0/go.mod h1:JxZsvoBklA50DNhELVJnJ567Rt+KrMH9rR3u515wvE8=
github.com/go-pkgz/lcw v0.5.0 h1:dSf98y5067zRmBA6fo+gbqLGLy08RGUHoaecwtjDyOE=
github.com/go-pkgz/lcw v0.5.0/go.mod h1:CSdQRQthxJQ4iDD4wTPPuWFbFdknJzwJ8WXu1nfxb10=
github.com/go-pkgz/lcw v0.5.1-0.20200509170726-dc283cfc28cf h1:lUztZWD77zDSILb3DHCpIpHxLILUmAf15aYNAEbWtIM=
github.com/go-pkgz/lcw v0.5.1-0.20200509170726-dc283cfc28cf/go.mod h1:vovP88gZLeuIWn5cm0NlgPYFyGGkv3m2OcKMOOaHhj0=
github.com/go-pkgz/lgr v0.7.0 h1:S/AAPwt/RE9a5mNJskA7dGVp+Dq6SMIW6LYjG3ITxY8=
github.com/go-pkgz/lgr v0.7.0/go.mod h1:yMgxU+GobMRJgIEbSzDKy/67W18S7qmGx/7BVL5AB8Q=
github.com/go-pkgz/repeater v1.1.3 h1:q6+JQF14ESSy28Dd7F+wRelY4F+41HJ0LEy/szNnMiE=
@@ -80,8 +80,8 @@ github.com/go-pkgz/rest v1.5.0 h1:C8SxXcXza4GiUUAn/95iCkvoIrGbS30qpwK19iqlrWQ=
github.com/go-pkgz/rest v1.5.0/go.mod h1:nQaM3RhSTUAmbBZWY4hfe4buyeC9VckvhoCktiQXJxI=
github.com/go-pkgz/syncs v1.1.1 h1:jWN+y6FS/Xe+8z4l3QMbSnODGyaxDHGojIS+wyKIjxg=
github.com/go-pkgz/syncs v1.1.1/go.mod h1:bt9lxWRRJ9vOCMGc8Big8ttjYHLKP88ofj1y38UlaHE=
github.com/go-redis/redis/v7 v7.0.0-beta.4 h1:p6z7Pde69EGRWvlC++y8aFcaWegyrKHzOBGo0zUACTQ=
github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
github.com/go-session/session v3.1.2+incompatible/go.mod h1:8B3iivBQjrz/JtC68Np2T1yBBLxTan3mn/3OM0CyRt0=
github.com/go-stack/stack v1.8.0 h1:5SgMzNM5HxrEjV0ww2lTmX6E2Izsfxas4+YHWRs3Lsk=
github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
@@ -97,8 +97,8 @@ github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3 h1:6amM4HsNPOvMLVc2ZnyqrjeQ92YAVWn7T4WBKK87inY=
github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
github.com/google/go-cmp v0.3.0 h1:crn/baboCvb5fXaQ0IJ1SGTsTVrWpDsCWC8EGETZijY=
@@ -116,16 +116,14 @@ github.com/gorilla/feeds v1.1.1 h1:HwKXxqzcRNg9to+BbvJog4+f3s/xzvtZXICcQGutYfY=
github.com/gorilla/feeds v1.1.1/go.mod h1:Nk0jZrvPFZX1OBe5NPiddPw7CfwF6Q9eqzaBbaightA=
github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI=
github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA=
github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.1 h1:0hERBMJE1eitiLkihrMvRVBYAkpHzc/J3QdDN+dAcgU=
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
github.com/hashicorp/golang-lru v0.5.3 h1:YPkqC67at8FYaadspW/6uE0COsBxS2656RLEr8Bppgk=
github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
github.com/imkira/go-interpol v1.1.0 h1:KIiKr0VSG2CUW1hl1jpiyuzuJeKUUpC8iM1AIE7N1Vk=
@@ -159,11 +157,11 @@ github.com/nullrocks/identicon v0.0.0-20180626043057-7875f45b0022 h1:Ys0rDzh8s4U
github.com/nullrocks/identicon v0.0.0-20180626043057-7875f45b0022/go.mod h1:x4NsS+uc7ecH/Cbm9xKQ6XzmJM57rWTkjywjfB2yQ18=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/onsi/gomega v1.5.0 h1:izbySO9zDPmjJ8rDjLvkA2zJHIo+HkYXHnf7eN7SSyo=
github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -235,8 +233,8 @@ github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 h1:BHyfKlQyqbsFN5p3Ifn
github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82/go.mod h1:lgjkn3NuSvDfVJdfcVVdX+jpBxNmX4rDAzaS45IcYoM=
github.com/yudai/pp v2.0.1+incompatible h1:Q4//iY4pNF6yPLZIigmvcl7k/bPgrcTPIFIcmawg5bI=
github.com/yudai/pp v2.0.1+incompatible/go.mod h1:PuxR/8QJ7cyCkFp/aUDS+JY727OFEZkTdatxwunjIkc=
github.com/yuin/gopher-lua v0.0.0-20190514113301-1cd887cd7036 h1:1b6PAtenNyhsmo/NKXVe34h7JEZKva1YB/ne7K7mqKM=
github.com/yuin/gopher-lua v0.0.0-20190514113301-1cd887cd7036/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb h1:ZkM6LRnq40pR1Ox0hTHlnpkcOTuFIDQpZ1IN8rKKhX0=
github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
go.etcd.io/bbolt v1.3.4 h1:hi1bXHMVrlQh6WwxAy+qZCV/SYIlqo+Ushwdpa4tAKg=
go.etcd.io/bbolt v1.3.4/go.mod h1:G5EMThwa9y8QZGBClrRx5EY+Yw9kAhnjy3bSjsnlVTQ=
go.mongodb.org/mongo-driver v1.1.1 h1:Sq1fR+0c58RME5EoqKdjkiQAmPjmfHlZOoRI6fTUOcs=
@@ -271,9 +269,9 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR5pz3Of3rY3CfYBVs4xY44aLks=
golang.org/x/net v0.0.0-20190611141213-3f473d35a33a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190724013045-ca1201d0de80 h1:Ao/3l156eZf2AW5wK8a7/smtodRU+gha3+BeqJ69lRk=
golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20200202094626-16171245cfb2 h1:CCH4IOTTfewWjGOlSp+zGcjutRKlBEZQ6wTn8ozI/nI=
golang.org/x/net v0.0.0-20200202094626-16171245cfb2/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -297,7 +295,7 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190507160741-ecd444e8653b/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190606165138-5da285871e9c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5 h1:LfCXLvNmTYH9kEmVgqbnsWfruoXZIrh4YBgqVHtDvw0=
golang.org/x/sys v0.0.0-20200202164722-d101bd2416d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
@@ -333,6 +331,8 @@ google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiq
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/oauth2.v3 v3.10.1 h1:/abis3O6tZFizY1/FgKGoDOmmu2ddvscBSSPdHVa6OI=
@@ -342,6 +342,8 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
+14 -10
View File
@@ -7,8 +7,6 @@ linters-settings:
min-complexity: 15
maligned:
suggest-new: true
dupl:
threshold: 100
goconst:
min-len: 2
min-occurrences: 2
@@ -25,9 +23,9 @@ linters-settings:
- wrapperFunc
linters:
disable-all: true
enable:
- megacheck
- golint
- govet
- unconvert
- megacheck
@@ -42,19 +40,25 @@ linters:
- typecheck
- ineffassign
- varcheck
- stylecheck
- gochecknoinits
- scopelint
- gocritic
- nakedret
- gosimple
- prealloc
fast: false
disable-all: true
run:
# modules-download-mode: vendor
output:
format: tab
skip-dirs:
- vendor
issues:
exclude-rules:
- text: "weak cryptographic primitive"
- text: "should have a package comment, unless it's in another file for this package"
linters:
- gosec
service:
golangci-lint-version: 1.19.x
- golint
exclude-use-default: false
+1 -1
View File
@@ -1,6 +1,6 @@
MIT License
Copyright (c) 2019 Umputun
Copyright (c) 2020 Umputun
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
+8 -7
View File
@@ -1,7 +1,6 @@
# Loading Cache Wrapper [![Build Status](https://github.com/go-pkgz/lcw/workflows/build/badge.svg)](https://github.com/go-pkgz/lcw/actions) [![Coverage Status](https://coveralls.io/repos/github/go-pkgz/lcw/badge.svg?branch=master)](https://coveralls.io/github/go-pkgz/lcw?branch=master) [![godoc](https://godoc.org/github.com/go-pkgz/lcw?status.svg)](https://godoc.org/github.com/go-pkgz/lcw)
The library adds a thin layer on top of [lru cache](https://github.com/hashicorp/golang-lru) and [patrickmn/go-cache](https://github.com/patrickmn/go-cache).
The library adds a thin layer on top of [lru cache](https://github.com/hashicorp/golang-lru) and internal implementation of expirable cache.
| Cache name | Constructor | Defaults | Description |
| -------------- | --------------------- | ----------------- | ----------------------- |
@@ -10,7 +9,6 @@ The library adds a thin layer on top of [lru cache](https://github.com/hashicorp
| RedisCache | lcw.NewRedisCache | ttl=5m | Redis cache with limits |
| Nop | lcw.NewNopCache | | Do-nothing cache |
Main features:
- LoadingCache (guava style)
- Limit maximum cache size (in bytes)
@@ -42,7 +40,6 @@ if err != nil {
}
s := val.(string) // cached value
```
### Cache with URI
@@ -58,14 +55,18 @@ Cache can be created with URIs:
`Scache` provides a wrapper on top of all implementations of `LoadingCache` with a number of special features:
1. Key is not a string but a composed type made from partition, key-id and list of scopes (tags).
1. Key is not a string, but a composed type made from partition, key-id and list of scopes (tags).
1. Value type limited to `[]byte`
1. Added `Flush` method for scoped/tagged invalidation of multiple records in a given partition
1. Simplified interface with Get, Stat and Flush only.
1. A simplified interface with Get, Stat and Flush only.
## Details
- In all cache types other than Redis (e.g. LRU and Expirable at the moment) values are stored as-is which means
that mutable values can be changed outside of cache. `ExampleLoadingCache_Mutability` illustrates that.
- All byte-size limits (MaxCacheSize and MaxValSize) only work for values implementing `lcw.Sizer` interface.
- Negative limits (max options) rejected
- `lgr.Value` wraps `interface{}` and should be converted back to the concrete type.
- The implementation started as a part of [remark42](https://github.com/umputun/remark) and later on moved to [go-pkgz/rest](https://github.com/go-pkgz/rest/tree/master/cache) library and finaly generalized to become `lcw`.
- The implementation started as a part of [remark42](https://github.com/umputun/remark)
and later on moved to [go-pkgz/rest](https://github.com/go-pkgz/rest/tree/master/cache)
library and finally generalized to become `lcw`.
+8 -2
View File
@@ -1,4 +1,4 @@
// Package lcw adds a thin layer on top of lru cache and go-cache providing more limits and common interface.
// Package lcw adds a thin layer on top of lru and expirable cache providing more limits and common interface.
// The primary method to get (and set) data to/from the cache is LoadingCache.Get returning stored data for a given key or
// call provided func to retrieve and store, similar to Guava loading cache.
// Limits allow max values for key size, number of keys, value size and total size of values in the cache.
@@ -28,6 +28,7 @@ type LoadingCache interface {
Purge() // clear cache
Stat() CacheStat // cache stats
Keys() []string // list of all keys
Close() error // close open connections
}
// CacheStat represent stats values
@@ -68,10 +69,15 @@ func (n *Nop) Purge() {}
// Delete does nothing for nop cache
func (n *Nop) Delete(key string) {}
// Delete does nothing for nop cache
// Keys does nothing for nop cache
func (n *Nop) Keys() []string { return nil }
// Stat always 0s for nop cache
func (n *Nop) Stat() CacheStat {
return CacheStat{}
}
// Close does nothing for nop cache
func (n *Nop) Close() error {
return nil
}
+35 -32
View File
@@ -4,8 +4,9 @@ import (
"sync/atomic"
"time"
cache "github.com/patrickmn/go-cache"
"github.com/pkg/errors"
"github.com/go-pkgz/lcw/internal/cache"
)
// ExpirableCache implements LoadingCache with TTL.
@@ -13,12 +14,11 @@ type ExpirableCache struct {
options
CacheStat
currentSize int64
backend *cache.Cache
backend *cache.LoadingCache
}
// NewExpirableCache makes expirable LoadingCache implementation, 1000 max keys by default and 5s TTL
// NewExpirableCache makes expirable LoadingCache implementation, 1000 max keys by default and 5m TTL
func NewExpirableCache(opts ...Option) (*ExpirableCache, error) {
res := ExpirableCache{
options: options{
maxKeys: 1000,
@@ -33,25 +33,30 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) {
}
}
res.backend = cache.New(res.ttl, res.ttl/2)
// OnEvicted called automatically for expired and manually deleted
res.backend.OnEvicted(func(key string, value interface{}) {
if res.onEvicted != nil {
res.onEvicted(key, value)
}
if s, ok := value.(Sizer); ok {
size := s.Size()
atomic.AddInt64(&res.currentSize, -1*int64(size))
}
})
backend, err := cache.NewLoadingCache(
cache.MaxKeys(res.maxKeys),
cache.TTL(res.ttl),
cache.PurgeEvery(res.ttl/2),
cache.OnEvicted(func(key string, value interface{}) {
if res.onEvicted != nil {
res.onEvicted(key, value)
}
if s, ok := value.(Sizer); ok {
size := s.Size()
atomic.AddInt64(&res.currentSize, -1*int64(size))
}
}),
)
if err != nil {
return nil, errors.Wrap(err, "error creating backend")
}
res.backend = backend
return &res, nil
}
// Get gets value by key or load with fn if not found in cache
func (c *ExpirableCache) Get(key string, fn func() (Value, error)) (data Value, err error) {
if v, ok := c.backend.Get(key); ok {
atomic.AddInt64(&c.Hits, 1)
return v, nil
@@ -71,7 +76,7 @@ func (c *ExpirableCache) Get(key string, fn func() (Value, error)) (data Value,
}
atomic.AddInt64(&c.currentSize, int64(s.Size()))
}
c.backend.Set(key, data, c.ttl)
c.backend.Set(key, data)
}
return data, nil
@@ -79,36 +84,28 @@ func (c *ExpirableCache) Get(key string, fn func() (Value, error)) (data Value,
// Invalidate removes keys with passed predicate fn, i.e. fn(key) should be true to get evicted
func (c *ExpirableCache) Invalidate(fn func(key string) bool) {
for key := range c.backend.Items() { // Keys() returns copy of cache's key, safe to remove directly
if fn(key) {
c.backend.Delete(key)
}
}
c.backend.InvalidateFn(fn)
}
// Peek returns the key value (or undefined if not found) without updating the "recently used"-ness of the key.
func (c *ExpirableCache) Peek(key string) (Value, bool) {
return c.backend.Get(key)
return c.backend.Peek(key)
}
// Purge clears the cache completely.
func (c *ExpirableCache) Purge() {
c.backend.Flush()
c.backend.Purge()
atomic.StoreInt64(&c.currentSize, 0)
}
// Delete cache item by key
func (c *ExpirableCache) Delete(key string) {
c.backend.Delete(key)
c.backend.Invalidate(key)
}
// Keys returns cache keys
func (c *ExpirableCache) Keys() (res []string) {
items := c.backend.Items()
res = make([]string, 0, len(items))
for key := range items {
res = append(res, key)
}
return res
return c.backend.Keys()
}
// Stat returns cache statistics
@@ -122,6 +119,12 @@ func (c *ExpirableCache) Stat() CacheStat {
}
}
// Close kills cleanup goroutine
func (c *ExpirableCache) Close() error {
c.backend.Close()
return nil
}
func (c *ExpirableCache) size() int64 {
return atomic.LoadInt64(&c.currentSize)
}
+6 -17
View File
@@ -1,23 +1,12 @@
module github.com/go-pkgz/lcw
require (
github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6 // indirect
github.com/alicebob/miniredis v2.5.0+incompatible
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/go-redis/redis/v7 v7.0.0-beta.4
github.com/golang/protobuf v1.3.2 // indirect
github.com/gomodule/redigo v2.0.0+incompatible // indirect
github.com/hashicorp/go-multierror v1.0.0
github.com/hashicorp/golang-lru v0.5.3
github.com/kr/pretty v0.1.0 // indirect
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/pkg/errors v0.8.1
github.com/stretchr/testify v1.3.0
github.com/yuin/gopher-lua v0.0.0-20190514113301-1cd887cd7036 // indirect
golang.org/x/net v0.0.0-20190620200207-3b0461eec859 // indirect
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0 // indirect
golang.org/x/text v0.3.2 // indirect
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
github.com/alicebob/miniredis/v2 v2.11.4
github.com/go-redis/redis/v7 v7.2.0
github.com/hashicorp/go-multierror v1.1.0
github.com/hashicorp/golang-lru v0.5.4
github.com/pkg/errors v0.9.1
github.com/stretchr/testify v1.5.1
)
go 1.13
+29 -38
View File
@@ -1,30 +1,27 @@
github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6 h1:45bxf7AZMwWcqkLzDAQugVEwedisr5nRJ1r+7LYnv0U=
github.com/alicebob/gopher-json v0.0.0-20180125190556-5a6b3ba71ee6/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
github.com/alicebob/miniredis v2.5.0+incompatible h1:yBHoLpsyjupjz3NL3MhKMVkR41j82Yjf3KFv7ApYzUI=
github.com/alicebob/miniredis v2.5.0+incompatible/go.mod h1:8HZjEj4yU0dwhYHky+DxYx+6BMjkBbe5ONFIF1MXffk=
github.com/alicebob/miniredis/v2 v2.11.4 h1:GsuyeunTx7EllZBU3/6Ji3dhMQZDpC9rLf1luJ+6M5M=
github.com/alicebob/miniredis/v2 v2.11.4/go.mod h1:VL3UDEfAH59bSa7MuHMuFToxkqyHh69s/WUbYlOAuyg=
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV9I=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/go-redis/redis/v7 v7.0.0-beta.4 h1:p6z7Pde69EGRWvlC++y8aFcaWegyrKHzOBGo0zUACTQ=
github.com/go-redis/redis/v7 v7.0.0-beta.4/go.mod h1:xhhSbUMTsleRPur+Vgx9sUHtyN33bdjxY+9/0n9Ig8s=
github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
github.com/go-redis/redis/v7 v7.2.0 h1:CrCexy/jYWZjW0AyVoHlcJUeZN19VWlbepTh1Vq6dJs=
github.com/go-redis/redis/v7 v7.2.0/go.mod h1:JDNMw23GTyLNC4GZu9njt15ctBQVn7xjRfnwdHj/Dcg=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
github.com/gomodule/redigo v2.0.0+incompatible/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3 h1:6amM4HsNPOvMLVc2ZnyqrjeQ92YAVWn7T4WBKK87inY=
github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
github.com/hashicorp/golang-lru v0.5.3 h1:YPkqC67at8FYaadspW/6uE0COsBxS2656RLEr8Bppgk=
github.com/hashicorp/golang-lru v0.5.3/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI=
github.com/hashicorp/go-multierror v1.1.0/go.mod h1:spPvp8C1qA32ftKqdAHm4hHTbPw+vmowP0z+KUhOZdA=
github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
@@ -33,47 +30,41 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.5.0 h1:izbySO9zDPmjJ8rDjLvkA2zJHIo+HkYXHnf7eN7SSyo=
github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/yuin/gopher-lua v0.0.0-20190514113301-1cd887cd7036 h1:1b6PAtenNyhsmo/NKXVe34h7JEZKva1YB/ne7K7mqKM=
github.com/yuin/gopher-lua v0.0.0-20190514113301-1cd887cd7036/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb h1:ZkM6LRnq40pR1Ox0hTHlnpkcOTuFIDQpZ1IN8rKKhX0=
github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd h1:nTDtHvHSdCn1m6ITfMRqtOd/9+7a3s8RBNOZ3eYZzJA=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e h1:o3PsSEY8E4eXWkXrIP9YJALUkVZqzHJT5DOasTyn8Vs=
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190204203706-41f3e6584952 h1:FDfvYgoVsA7TTZSbgiqjAbfPbK47CNHdWl3h/PJtii0=
golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0 h1:HyfiK1WMnHj5FXFXatD+Qs1A/xC2Run6RzeW1SyHxpc=
golang.org/x/sys v0.0.0-20190624142023-c5567b49c5d0/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+242
View File
@@ -0,0 +1,242 @@
// Package cache implements LoadingCache.
//
// Support LRC TTL-based eviction.
package cache
import (
"sort"
"sync"
"time"
"github.com/pkg/errors"
)
// LoadingCache provides expirable loading cache with LRC eviction.
type LoadingCache struct {
purgeEvery time.Duration
ttl time.Duration
maxKeys int64
done chan struct{}
onEvicted func(key string, value interface{})
sync.Mutex
data map[string]*cacheItem
}
// noEvictionTTL - very long ttl to prevent eviction
const noEvictionTTL = time.Hour * 24 * 365 * 10
// NewLoadingCache returns a new expirable LRC cache, activates purge with purgeEvery (0 to never purge).
// Default MaxKeys is unlimited (0).
func NewLoadingCache(options ...Option) (*LoadingCache, error) {
res := LoadingCache{
data: map[string]*cacheItem{},
ttl: noEvictionTTL,
purgeEvery: 0,
maxKeys: 0,
done: make(chan struct{}),
}
for _, opt := range options {
if err := opt(&res); err != nil {
return nil, errors.Wrap(err, "failed to set cache option")
}
}
if res.maxKeys > 0 || res.purgeEvery > 0 {
if res.purgeEvery == 0 {
res.purgeEvery = time.Minute * 5 // non-zero purge enforced because maxKeys defined
}
go func(done <-chan struct{}) {
ticker := time.NewTicker(res.purgeEvery)
for {
select {
case <-done:
return
case <-ticker.C:
res.Lock()
res.purge(res.maxKeys)
res.Unlock()
}
}
}(res.done)
}
return &res, nil
}
// Set key
func (c *LoadingCache) Set(key string, value interface{}) {
c.Lock()
defer c.Unlock()
now := time.Now()
if _, ok := c.data[key]; !ok {
c.data[key] = &cacheItem{}
}
c.data[key].data = value
c.data[key].expiresAt = now.Add(c.ttl)
// Enforced purge call in addition the one from the ticker
// to limit the worst-case scenario with a lot of sets in the
// short period of time (between two timed purge calls)
if c.maxKeys > 0 && int64(len(c.data)) >= c.maxKeys*2 {
c.purge(c.maxKeys)
}
}
// Get returns the key value
func (c *LoadingCache) Get(key string) (interface{}, bool) {
c.Lock()
defer c.Unlock()
value, ok := c.getValue(key)
if !ok {
return nil, false
}
return value, ok
}
// Peek returns the key value (or undefined if not found) without updating the "recently used"-ness of the key.
func (c *LoadingCache) Peek(key string) (interface{}, bool) {
c.Lock()
defer c.Unlock()
value, ok := c.getValue(key)
if !ok {
return nil, false
}
return value, ok
}
// Invalidate key (item) from the cache
func (c *LoadingCache) Invalidate(key string) {
c.Lock()
if value, ok := c.data[key]; ok {
delete(c.data, key)
if c.onEvicted != nil {
c.onEvicted(key, value.data)
}
}
c.Unlock()
}
// InvalidateFn deletes multiple keys if predicate is true
func (c *LoadingCache) InvalidateFn(fn func(key string) bool) {
c.Lock()
for key, value := range c.data {
if fn(key) {
delete(c.data, key)
if c.onEvicted != nil {
c.onEvicted(key, value.data)
}
}
}
c.Unlock()
}
// Keys return slice of current keys in the cache
func (c *LoadingCache) Keys() []string {
c.Lock()
defer c.Unlock()
keys := make([]string, 0, len(c.data))
for k := range c.data {
keys = append(keys, k)
}
return keys
}
// get value respecting the expiration, should be called with lock
func (c *LoadingCache) getValue(key string) (interface{}, bool) {
value, ok := c.data[key]
if !ok {
return nil, false
}
if time.Now().After(c.data[key].expiresAt) {
return nil, false
}
return value.data, ok
}
// Purge clears the cache completely.
func (c *LoadingCache) Purge() {
c.Lock()
defer c.Unlock()
for k, v := range c.data {
delete(c.data, k)
if c.onEvicted != nil {
c.onEvicted(k, v.data)
}
}
}
// DeleteExpired clears cache of expired items
func (c *LoadingCache) DeleteExpired() {
c.Lock()
defer c.Unlock()
c.purge(0)
}
// ItemCount return count of items in cache
func (c *LoadingCache) ItemCount() int {
c.Lock()
n := len(c.data)
c.Unlock()
return n
}
// Close cleans the cache and destroys running goroutines
func (c *LoadingCache) Close() {
c.Lock()
defer c.Unlock()
close(c.done)
}
// keysWithTs includes list of keys with ts. This is for sorting keys
// in order to provide least recently added sorting for size-based eviction
type keysWithTs []struct {
key string
ts time.Time
}
// purge records > maxKeys. Has to be called with lock!
// call with maxKeys 0 will only clear expired entries.
func (c *LoadingCache) purge(maxKeys int64) {
kts := keysWithTs{}
for key, value := range c.data {
// ttl eviction
if time.Now().After(c.data[key].expiresAt) {
delete(c.data, key)
if c.onEvicted != nil {
c.onEvicted(key, value.data)
}
}
// prepare list of keysWithTs for size eviction
if maxKeys > 0 && int64(len(c.data)) > maxKeys {
ts := c.data[key].expiresAt
kts = append(kts, struct {
key string
ts time.Time
}{key, ts})
}
}
// size eviction
size := int64(len(c.data))
if len(kts) > 0 {
sort.Slice(kts, func(i int, j int) bool { return kts[i].ts.Before(kts[j].ts) })
for d := 0; int64(d) < size-maxKeys; d++ {
key := kts[d].key
value := c.data[key].data
delete(c.data, key)
if c.onEvicted != nil {
c.onEvicted(key, value)
}
}
}
}
type cacheItem struct {
expiresAt time.Time
data interface{}
}
+42
View File
@@ -0,0 +1,42 @@
package cache
import "time"
// Option func type
type Option func(lc *LoadingCache) error
// OnEvicted called automatically for expired and manually deleted entries
func OnEvicted(fn func(key string, value interface{})) Option {
return func(lc *LoadingCache) error {
lc.onEvicted = fn
return nil
}
}
// PurgeEvery functional option defines purge interval
// by default it is 0, i.e. never. If MaxKeys set to any non-zero this default will be 5minutes
func PurgeEvery(interval time.Duration) Option {
return func(lc *LoadingCache) error {
lc.purgeEvery = interval
return nil
}
}
// MaxKeys functional option defines how many keys to keep.
// By default it is 0, which means unlimited.
// If any non-zero MaxKeys set, default PurgeEvery will be set to 5 minutes
func MaxKeys(max int) Option {
return func(lc *LoadingCache) error {
lc.maxKeys = int64(max)
return nil
}
}
// TTL functional option defines TTL for all cache entries.
// By default it is set to 10 years, sane option for expirable cache might be 5 minutes.
func TTL(ttl time.Duration) Option {
return func(lc *LoadingCache) error {
lc.ttl = ttl
return nil
}
}
+6 -2
View File
@@ -17,7 +17,6 @@ type LruCache struct {
// NewLruCache makes LRU LoadingCache implementation, 1000 max keys by default
func NewLruCache(opts ...Option) (*LruCache, error) {
res := LruCache{
options: options{
maxKeys: 1000,
@@ -51,7 +50,6 @@ func NewLruCache(opts ...Option) (*LruCache, error) {
// Get gets value by key or load with fn if not found in cache
func (c *LruCache) Get(key string, fn func() (Value, error)) (data Value, err error) {
if v, ok := c.backend.Get(key); ok {
atomic.AddInt64(&c.Hits, 1)
return v, nil
@@ -104,6 +102,7 @@ func (c *LruCache) Delete(key string) {
c.backend.Remove(key)
}
// Keys returns cache keys
func (c *LruCache) Keys() (res []string) {
keys := c.backend.Keys()
res = make([]string, 0, len(keys))
@@ -124,6 +123,11 @@ func (c *LruCache) Stat() CacheStat {
}
}
// Close does nothing for this type of cache
func (c *LruCache) Close() error {
return nil
}
func (c *LruCache) size() int64 {
return atomic.LoadInt64(&c.currentSize)
}
+6 -4
View File
@@ -4,7 +4,7 @@ import (
"sync/atomic"
"time"
redis "github.com/go-redis/redis/v7"
"github.com/go-redis/redis/v7"
"github.com/pkg/errors"
)
@@ -20,7 +20,6 @@ type RedisCache struct {
// NewRedisCache makes Redis LoadingCache implementation.
func NewRedisCache(backend *redis.Client, opts ...Option) (*RedisCache, error) {
res := RedisCache{
options: options{
ttl: 5 * time.Minute,
@@ -43,7 +42,6 @@ func NewRedisCache(backend *redis.Client, opts ...Option) (*RedisCache, error) {
// Get gets value by key or load with fn if not found in cache
func (c *RedisCache) Get(key string, fn func() (Value, error)) (data Value, err error) {
v, getErr := c.backend.Get(key).Result()
switch getErr {
// RedisClient returns nil when find a key in DB
@@ -69,7 +67,6 @@ func (c *RedisCache) Get(key string, fn func() (Value, error)) (data Value, err
atomic.AddInt64(&c.Errors, 1)
return data, setErr
}
}
return data, nil
}
@@ -119,6 +116,11 @@ func (c *RedisCache) Stat() CacheStat {
}
}
// Close closes underlying connections
func (c *RedisCache) Close() error {
return c.backend.Close()
}
func (c *RedisCache) size() int64 {
return 0
}
+6 -6
View File
@@ -17,6 +17,7 @@ func NewScache(lc LoadingCache) *Scache {
return &Scache{lc: lc}
}
// Get retrieves a key from underlying backend
func (m *Scache) Get(key Key, fn func() ([]byte, error)) (data []byte, err error) {
keyStr := key.String()
val, err := m.lc.Get(keyStr, func() (value Value, e error) {
@@ -32,7 +33,6 @@ func (m *Scache) Stat() CacheStat {
// Flush clears cache and calls postFlushFn async
func (m *Scache) Flush(req FlusherRequest) {
if len(req.scopes) == 0 {
m.lc.Purge()
return
@@ -93,11 +93,11 @@ func (k Key) Scopes(scopes ...string) Key {
// key string made as <partition>@@<id>@@<scope1>$$<scope2>....
func (k Key) String() string {
bld := strings.Builder{}
bld.WriteString(k.partition)
bld.WriteString("@@")
bld.WriteString(k.id)
bld.WriteString("@@")
bld.WriteString(strings.Join(k.scopes, "$$"))
_, _ = bld.WriteString(k.partition)
_, _ = bld.WriteString("@@")
_, _ = bld.WriteString(k.id)
_, _ = bld.WriteString("@@")
_, _ = bld.WriteString(strings.Join(k.scopes, "$$"))
return bld.String()
}
-1
View File
@@ -52,7 +52,6 @@ func New(uri string) (LoadingCache, error) {
}
func optionsFromQuery(q url.Values) (opts []Option, err error) {
errs := new(multierror.Error)
if v := q.Get("max_val_size"); v != "" {
+4
View File
@@ -5,7 +5,11 @@ run:
linters:
enable-all: true
disable:
- funlen
- gochecknoglobals
- gocognit
- goconst
- godox
- gosec
- maligned
- wsl
+2 -1
View File
@@ -8,6 +8,7 @@ services:
go:
- 1.11.x
- 1.12.x
- 1.13.x
- tip
matrix:
@@ -20,4 +21,4 @@ env:
go_import_path: github.com/go-redis/redis
before_install:
- curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $(go env GOPATH)/bin v1.17.1
- curl -sfL https://install.goreleaser.com/github.com/golangci/golangci-lint.sh | sh -s -- -b $(go env GOPATH)/bin v1.21.0
+9 -2
View File
@@ -1,14 +1,21 @@
# Changelog
## v7 WIP
## v7.2
- Existing `HMSet` is renamed to `HSet` and old deprecated `HMSet` is restored for Redis 3 users.
## v7
- *Important*. Tx.Pipeline now returns a non-transactional pipeline. Use Tx.TxPipeline for a transactional pipeline.
- WrapProcess is replaced with more convenient AddHook that has access to context.Context.
- WithContext now can not be used to create a shallow copy of the client.
- New methods ProcessContext, DoContext, and ExecContext.
- Client respects Context.Deadline when setting net.Conn deadline.
- Client listens on Context.Done while waiting for a connection from the pool and returns an error when context context is cancelled.
- Add PubSub.ChannelWithSubscriptions that sends `*Subscription` in addition to `*Message` to allow detecting reconnections.
- `time.Time` is now marshalled in RFC3339 format. `rdb.Get("foo").Time()` helper is added to parse time.
- `time.Time` is now marshalled in RFC3339 format. `rdb.Get("foo").Time()` helper is added to parse the time.
- `SetLimiter` is removed and added `Options.Limiter` instead.
- `HMSet` is deprecated as of Redis v4.
## v6.15
+1 -5
View File
@@ -3,9 +3,6 @@ all: testdeps
go test ./... -short -race
go test ./... -run=NONE -bench=. -benchmem
env GOOS=linux GOARCH=386 go test ./...
go vet
go get github.com/gordonklaus/ineffassign
ineffassign .
golangci-lint run
testdeps: testdata/redis/src/redis-server
@@ -17,8 +14,7 @@ bench: testdeps
testdata/redis:
mkdir -p $@
wget -qO- https://github.com/antirez/redis/archive/5.0.tar.gz | tar xvz --strip-components=1 -C $@
wget -qO- http://download.redis.io/releases/redis-5.0.7.tar.gz | tar xvz --strip-components=1 -C $@
testdata/redis/src/redis-server: testdata/redis
sed -i.bak 's/libjemalloc.a/libjemalloc.a -lrt/g' $</src/Makefile
cd $< && make all
+15 -9
View File
@@ -27,21 +27,22 @@ Examples: https://godoc.org/github.com/go-redis/redis#pkg-examples.
## Installation
Install:
go-redis requires a Go version with [Modules](https://github.com/golang/go/wiki/Modules) support and uses import versioning. So please make sure to initialize a Go module before installing go-redis:
```shell
go get -u github.com/go-redis/redis
``` shell
go mod init github.com/my/repo
go get github.com/go-redis/redis/v7
```
Import:
```go
import "github.com/go-redis/redis"
``` go
import "github.com/go-redis/redis/v7"
```
## Quickstart
```go
``` go
func ExampleNewClient() {
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
@@ -55,6 +56,11 @@ func ExampleNewClient() {
}
func ExampleClient() {
client := redis.NewClient(&redis.Options{
Addr: "localhost:6379",
Password: "", // no password set
DB: 0, // use default DB
})
err := client.Set("key", "value", 0).Err()
if err != nil {
panic(err)
@@ -87,12 +93,12 @@ Please go through [examples](https://godoc.org/github.com/go-redis/redis#pkg-exa
Some corner cases:
```go
``` go
// SET key value EX 10 NX
set, err := client.SetNX("key", "value", 10*time.Second).Result()
// SORT list LIMIT 0 2 ASC
vals, err := client.Sort("list", redis.Sort{Offset: 0, Count: 2, Order: "ASC"}).Result()
vals, err := client.Sort("list", &redis.Sort{Offset: 0, Count: 2, Order: "ASC"}).Result()
// ZRANGEBYSCORE zset -inf +inf WITHSCORES LIMIT 0 2
vals, err := client.ZRangeByScoreWithScores("zset", redis.ZRangeBy{
@@ -109,7 +115,7 @@ vals, err := client.ZInterStore("out", redis.ZStore{Weights: []int64{2, 3}}, "zs
vals, err := client.Eval("return {KEYS[1],ARGV[1]}", []string{"key"}, "hello").Result()
// custom command
res, err := client.Do("set", "key", "value")
res, err := client.Do("set", "key", "value").Result()
```
## See also
+126 -90
View File
@@ -696,7 +696,7 @@ func (c *ClusterClient) WithContext(ctx context.Context) *ClusterClient {
}
clone := *c
clone.cmdable = clone.Process
clone.hooks.Lock()
clone.hooks.lock()
clone.ctx = ctx
return &clone
}
@@ -743,7 +743,7 @@ func (c *ClusterClient) ProcessContext(ctx context.Context, cmd Cmder) error {
func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
err := c._process(ctx, cmd)
if err != nil {
cmd.setErr(err)
cmd.SetErr(err)
return err
}
return nil
@@ -773,13 +773,13 @@ func (c *ClusterClient) _process(ctx context.Context, cmd Cmder) error {
if ask {
pipe := node.Client.Pipeline()
_ = pipe.Process(NewCmd("ASKING"))
_ = pipe.Process(NewCmd("asking"))
_ = pipe.Process(cmd)
_, lastErr = pipe.ExecContext(ctx)
_ = pipe.Close()
ask = false
} else {
lastErr = node.Client._process(ctx, cmd)
lastErr = node.Client.ProcessContext(ctx, cmd)
}
// If there is no error - we are done.
@@ -840,6 +840,7 @@ func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, master := range state.Masters {
wg.Add(1)
go func(node *clusterNode) {
@@ -853,6 +854,7 @@ func (c *ClusterClient) ForEachMaster(fn func(client *Client) error) error {
}
}(master)
}
wg.Wait()
select {
@@ -873,6 +875,7 @@ func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error {
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, slave := range state.Slaves {
wg.Add(1)
go func(node *clusterNode) {
@@ -886,6 +889,7 @@ func (c *ClusterClient) ForEachSlave(fn func(client *Client) error) error {
}
}(slave)
}
wg.Wait()
select {
@@ -906,6 +910,7 @@ func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error {
var wg sync.WaitGroup
errCh := make(chan error, 1)
worker := func(node *clusterNode) {
defer wg.Done()
err := fn(node.Client)
@@ -927,6 +932,7 @@ func (c *ClusterClient) ForEachNode(fn func(client *Client) error) error {
}
wg.Wait()
select {
case err := <-errCh:
return err
@@ -1046,7 +1052,7 @@ func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error
func (c *ClusterClient) _processPipeline(ctx context.Context, cmds []Cmder) error {
cmdsMap := newCmdsMap()
err := c.mapCmdsByNode(cmds, cmdsMap)
err := c.mapCmdsByNode(cmdsMap, cmds)
if err != nil {
setCmdsErr(cmds, err)
return err
@@ -1068,23 +1074,16 @@ func (c *ClusterClient) _processPipeline(ctx context.Context, cmds []Cmder) erro
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
err := node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmds...)
})
if err != nil {
return err
}
return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
return c.pipelineReadCmds(node, rd, cmds, failedCmds)
})
})
if err != nil {
err = c.mapCmdsByNode(cmds, failedCmds)
if err != nil {
err := c._processPipelineNode(ctx, node, cmds, failedCmds)
if err == nil {
return
}
if attempt < c.opt.MaxRedirects {
if err := c.mapCmdsByNode(failedCmds, cmds); err != nil {
setCmdsErr(cmds, err)
}
} else {
setCmdsErr(cmds, err)
}
}(node, cmds)
}
@@ -1099,41 +1098,27 @@ func (c *ClusterClient) _processPipeline(ctx context.Context, cmds []Cmder) erro
return cmdsFirstErr(cmds)
}
type cmdsMap struct {
mu sync.Mutex
m map[*clusterNode][]Cmder
}
func newCmdsMap() *cmdsMap {
return &cmdsMap{
m: make(map[*clusterNode][]Cmder),
}
}
func (m *cmdsMap) Add(node *clusterNode, cmds ...Cmder) {
m.mu.Lock()
m.m[node] = append(m.m[node], cmds...)
m.mu.Unlock()
}
func (c *ClusterClient) mapCmdsByNode(cmds []Cmder, cmdsMap *cmdsMap) error {
func (c *ClusterClient) mapCmdsByNode(cmdsMap *cmdsMap, cmds []Cmder) error {
state, err := c.state.Get()
if err != nil {
return err
}
cmdsAreReadOnly := c.opt.ReadOnly && c.cmdsAreReadOnly(cmds)
if c.opt.ReadOnly && c.cmdsAreReadOnly(cmds) {
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
node, err := c.slotReadOnlyNode(state, slot)
if err != nil {
return err
}
cmdsMap.Add(node, cmd)
}
return nil
}
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
var node *clusterNode
var err error
if cmdsAreReadOnly {
cmdInfo := c.cmdInfo(cmd.Name())
node, err = c.cmdNode(cmdInfo, slot)
} else {
node, err = state.slotMasterNode(slot)
}
node, err := state.slotMasterNode(slot)
if err != nil {
return err
}
@@ -1152,6 +1137,25 @@ func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
return true
}
func (c *ClusterClient) _processPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) error {
return node.Client.hooks.processPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmds(wr, cmds)
})
if err != nil {
return err
}
return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
return c.pipelineReadCmds(node, rd, cmds, failedCmds)
})
})
})
}
func (c *ClusterClient) pipelineReadCmds(
node *clusterNode, rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
) error {
@@ -1196,7 +1200,7 @@ func (c *ClusterClient) checkMovedErr(
}
if ask {
failedCmds.Add(node, NewCmd("ASKING"), cmd)
failedCmds.Add(node, NewCmd("asking"), cmd)
return true
}
@@ -1253,32 +1257,16 @@ func (c *ClusterClient) _processTxPipeline(ctx context.Context, cmds []Cmder) er
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
err := node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return txPipelineWriteMulti(wr, cmds)
})
if err != nil {
return err
}
err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
err := c.txPipelineReadQueued(rd, cmds, failedCmds)
if err != nil {
moved, ask, addr := isMovedError(err)
if moved || ask {
return c.cmdsMoved(cmds, moved, ask, addr, failedCmds)
}
return err
}
return pipelineReadCmds(rd, cmds)
})
return err
})
if err != nil {
err = c.mapCmdsByNode(cmds, failedCmds)
if err != nil {
err := c._processTxPipelineNode(ctx, node, cmds, failedCmds)
if err == nil {
return
}
if attempt < c.opt.MaxRedirects {
if err := c.mapCmdsByNode(failedCmds, cmds); err != nil {
setCmdsErr(cmds, err)
}
} else {
setCmdsErr(cmds, err)
}
}(node, cmds)
}
@@ -1303,11 +1291,42 @@ func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
return cmdsMap
}
func (c *ClusterClient) _processTxPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) error {
return node.Client.hooks.processTxPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmds(wr, cmds)
})
if err != nil {
return err
}
return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
cmds = cmds[1 : len(cmds)-1]
err := c.txPipelineReadQueued(rd, statusCmd, cmds, failedCmds)
if err != nil {
moved, ask, addr := isMovedError(err)
if moved || ask {
return c.cmdsMoved(cmds, moved, ask, addr, failedCmds)
}
return err
}
return pipelineReadCmds(rd, cmds)
})
})
})
}
func (c *ClusterClient) txPipelineReadQueued(
rd *proto.Reader, cmds []Cmder, failedCmds *cmdsMap,
rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder, failedCmds *cmdsMap,
) error {
// Parse queued replies.
var statusCmd StatusCmd
if err := statusCmd.readReply(rd); err != nil {
return err
}
@@ -1359,7 +1378,7 @@ func (c *ClusterClient) cmdsMoved(
if ask {
for _, cmd := range cmds {
failedCmds.Add(node, NewCmd("ASKING"), cmd)
failedCmds.Add(node, NewCmd("asking"), cmd)
}
return nil
}
@@ -1561,29 +1580,27 @@ func (c *ClusterClient) cmdNode(cmdInfo *CommandInfo, slot int) (*clusterNode, e
}
if c.opt.ReadOnly && cmdInfo != nil && cmdInfo.ReadOnly {
if c.opt.RouteByLatency {
return state.slotClosestNode(slot)
}
if c.opt.RouteRandomly {
return state.slotRandomNode(slot)
}
return state.slotSlaveNode(slot)
return c.slotReadOnlyNode(state, slot)
}
return state.slotMasterNode(slot)
}
func (c *clusterClient) slotReadOnlyNode(state *clusterState, slot int) (*clusterNode, error) {
if c.opt.RouteByLatency {
return state.slotClosestNode(slot)
}
if c.opt.RouteRandomly {
return state.slotRandomNode(slot)
}
return state.slotSlaveNode(slot)
}
func (c *ClusterClient) slotMasterNode(slot int) (*clusterNode, error) {
state, err := c.state.Get()
if err != nil {
return nil, err
}
nodes := state.slotNodes(slot)
if len(nodes) > 0 {
return nodes[0], nil
}
return c.nodes.Random()
return state.slotMasterNode(slot)
}
func appendUniqueNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
@@ -1622,3 +1639,22 @@ func remove(ss []string, es ...string) []string {
}
return ss
}
//------------------------------------------------------------------------------
type cmdsMap struct {
mu sync.Mutex
m map[*clusterNode][]Cmder
}
func newCmdsMap() *cmdsMap {
return &cmdsMap{
m: make(map[*clusterNode][]Cmder),
}
}
func (m *cmdsMap) Add(node *clusterNode, cmds ...Cmder) {
m.mu.Lock()
m.m[node] = append(m.m[node], cmds...)
m.mu.Unlock()
}
+1 -1
View File
@@ -14,7 +14,7 @@ func (c *ClusterClient) DBSize() *IntCmd {
return nil
})
if err != nil {
cmd.setErr(err)
cmd.SetErr(err)
return cmd
}
cmd.val = size
+126 -10
View File
@@ -15,19 +15,20 @@ import (
type Cmder interface {
Name() string
Args() []interface{}
String() string
stringArg(int) string
readTimeout() *time.Duration
readReply(rd *proto.Reader) error
setErr(error)
SetErr(error)
Err() error
}
func setCmdsErr(cmds []Cmder, e error) {
for _, cmd := range cmds {
if cmd.Err() == nil {
cmd.setErr(e)
cmd.SetErr(e)
}
}
}
@@ -41,16 +42,19 @@ func cmdsFirstErr(cmds []Cmder) error {
return nil
}
func writeCmd(wr *proto.Writer, cmds ...Cmder) error {
func writeCmds(wr *proto.Writer, cmds []Cmder) error {
for _, cmd := range cmds {
err := wr.WriteArgs(cmd.Args())
if err != nil {
if err := writeCmd(wr, cmd); err != nil {
return err
}
}
return nil
}
func writeCmd(wr *proto.Writer, cmd Cmder) error {
return wr.WriteArgs(cmd.Args())
}
func cmdString(cmd Cmder, val interface{}) string {
ss := make([]string, 0, len(cmd.Args()))
for _, arg := range cmd.Args() {
@@ -119,7 +123,7 @@ func (cmd *baseCmd) stringArg(pos int) string {
return s
}
func (cmd *baseCmd) setErr(e error) {
func (cmd *baseCmd) SetErr(e error) {
cmd.err = e
}
@@ -149,6 +153,10 @@ func NewCmd(args ...interface{}) *Cmd {
}
}
func (cmd *Cmd) String() string {
return cmdString(cmd, cmd.val)
}
func (cmd *Cmd) Val() interface{} {
return cmd.val
}
@@ -157,7 +165,7 @@ func (cmd *Cmd) Result() (interface{}, error) {
return cmd.val, cmd.err
}
func (cmd *Cmd) String() (string, error) {
func (cmd *Cmd) Text() (string, error) {
if cmd.err != nil {
return "", cmd.err
}
@@ -385,6 +393,10 @@ func (cmd *IntCmd) Result() (int64, error) {
return cmd.val, cmd.err
}
func (cmd *IntCmd) Uint64() (uint64, error) {
return uint64(cmd.val), cmd.err
}
func (cmd *IntCmd) String() string {
return cmdString(cmd, cmd.val)
}
@@ -999,14 +1011,20 @@ func xMessageSliceParser(rd *proto.Reader, n int64) (interface{}, error) {
return nil, err
}
var values map[string]interface{}
v, err := rd.ReadArrayReply(stringInterfaceMapParser)
if err != nil {
return nil, err
if err != proto.Nil {
return nil, err
}
} else {
values = v.(map[string]interface{})
}
msgs[i] = XMessage{
ID: id,
Values: v.(map[string]interface{}),
Values: values,
}
return nil, nil
})
@@ -1285,6 +1303,96 @@ func (cmd *XPendingExtCmd) readReply(rd *proto.Reader) error {
//------------------------------------------------------------------------------
type XInfoGroupsCmd struct {
baseCmd
val []XInfoGroups
}
type XInfoGroups struct {
Name string
Consumers int64
Pending int64
LastDeliveredID string
}
var _ Cmder = (*XInfoGroupsCmd)(nil)
func NewXInfoGroupsCmd(stream string) *XInfoGroupsCmd {
return &XInfoGroupsCmd{
baseCmd: baseCmd{args: []interface{}{"xinfo", "groups", stream}},
}
}
func (cmd *XInfoGroupsCmd) Val() []XInfoGroups {
return cmd.val
}
func (cmd *XInfoGroupsCmd) Result() ([]XInfoGroups, error) {
return cmd.val, cmd.err
}
func (cmd *XInfoGroupsCmd) String() string {
return cmdString(cmd, cmd.val)
}
func (cmd *XInfoGroupsCmd) readReply(rd *proto.Reader) error {
_, cmd.err = rd.ReadArrayReply(
func(rd *proto.Reader, n int64) (interface{}, error) {
for i := int64(0); i < n; i++ {
v, err := rd.ReadReply(xGroupInfoParser)
if err != nil {
return nil, err
}
cmd.val = append(cmd.val, v.(XInfoGroups))
}
return nil, nil
})
return nil
}
func xGroupInfoParser(rd *proto.Reader, n int64) (interface{}, error) {
if n != 8 {
return nil, fmt.Errorf("redis: got %d elements in XINFO GROUPS reply,"+
"wanted 8", n)
}
var (
err error
grp XInfoGroups
key string
val string
)
for i := 0; i < 4; i++ {
key, err = rd.ReadString()
if err != nil {
return nil, err
}
val, err = rd.ReadString()
if err != nil {
return nil, err
}
switch key {
case "name":
grp.Name = val
case "consumers":
grp.Consumers, err = strconv.ParseInt(val, 0, 64)
case "pending":
grp.Pending, err = strconv.ParseInt(val, 0, 64)
case "last-delivered-id":
grp.LastDeliveredID = val
default:
return nil, fmt.Errorf("redis: unexpected content %s "+
"in XINFO GROUPS reply", key)
}
if err != nil {
return nil, err
}
}
return grp, err
}
//------------------------------------------------------------------------------
type ZSliceCmd struct {
baseCmd
@@ -1769,7 +1877,6 @@ func (cmd *GeoPosCmd) readReply(rd *proto.Reader) error {
return nil, nil
})
return cmd.err
}
//------------------------------------------------------------------------------
@@ -1912,6 +2019,15 @@ func (c *cmdsInfoCache) Get() (map[string]*CommandInfo, error) {
if err != nil {
return err
}
// Extensions have cmd names in upper case. Convert them to lower case.
for k, v := range cmds {
lower := internal.ToLower(k)
if lower != k {
cmds[lower] = v
}
}
c.cmds = cmds
return nil
})
+60 -30
View File
@@ -34,11 +34,17 @@ func formatSec(dur time.Duration) int64 {
func appendArgs(dst, src []interface{}) []interface{} {
if len(src) == 1 {
if ss, ok := src[0].([]string); ok {
for _, s := range ss {
switch v := src[0].(type) {
case []string:
for _, s := range v {
dst = append(dst, s)
}
return dst
case map[string]interface{}:
for k, v := range v {
dst = append(dst, k, v)
}
return dst
}
}
@@ -107,8 +113,8 @@ type Cmdable interface {
IncrBy(key string, value int64) *IntCmd
IncrByFloat(key string, value float64) *FloatCmd
MGet(keys ...string) *SliceCmd
MSet(pairs ...interface{}) *StatusCmd
MSetNX(pairs ...interface{}) *BoolCmd
MSet(values ...interface{}) *StatusCmd
MSetNX(values ...interface{}) *BoolCmd
Set(key string, value interface{}, expiration time.Duration) *StatusCmd
SetBit(key string, offset int64, value int) *IntCmd
SetNX(key string, value interface{}, expiration time.Duration) *BoolCmd
@@ -124,8 +130,8 @@ type Cmdable interface {
HKeys(key string) *StringSliceCmd
HLen(key string) *IntCmd
HMGet(key string, fields ...string) *SliceCmd
HMSet(key string, fields map[string]interface{}) *StatusCmd
HSet(key, field string, value interface{}) *BoolCmd
HSet(key string, values ...interface{}) *IntCmd
HMSet(key string, values ...interface{}) *BoolCmd
HSetNX(key, field string, value interface{}) *BoolCmd
HVals(key string) *StringSliceCmd
BLPop(timeout time.Duration, keys ...string) *StringSliceCmd
@@ -186,6 +192,7 @@ type Cmdable interface {
XClaimJustID(a *XClaimArgs) *StringSliceCmd
XTrim(key string, maxLen int64) *IntCmd
XTrimApprox(key string, maxLen int64) *IntCmd
XInfoGroups(key string) *XInfoGroupsCmd
BZPopMax(timeout time.Duration, keys ...string) *ZWithKeyCmd
BZPopMin(timeout time.Duration, keys ...string) *ZWithKeyCmd
ZAdd(key string, members ...*Z) *IntCmd
@@ -799,19 +806,27 @@ func (c cmdable) MGet(keys ...string) *SliceCmd {
return cmd
}
func (c cmdable) MSet(pairs ...interface{}) *StatusCmd {
args := make([]interface{}, 1, 1+len(pairs))
// MSet is like Set but accepts multiple values:
// - MSet("key1", "value1", "key2", "value2")
// - MSet([]string{"key1", "value1", "key2", "value2"})
// - MSet(map[string]interface{}{"key1": "value1", "key2": "value2"})
func (c cmdable) MSet(values ...interface{}) *StatusCmd {
args := make([]interface{}, 1, 1+len(values))
args[0] = "mset"
args = appendArgs(args, pairs)
args = appendArgs(args, values)
cmd := NewStatusCmd(args...)
_ = c(cmd)
return cmd
}
func (c cmdable) MSetNX(pairs ...interface{}) *BoolCmd {
args := make([]interface{}, 1, 1+len(pairs))
// MSetNX is like SetNX but accepts multiple values:
// - MSetNX("key1", "value1", "key2", "value2")
// - MSetNX([]string{"key1", "value1", "key2", "value2"})
// - MSetNX(map[string]interface{}{"key1": "value1", "key2": "value2"})
func (c cmdable) MSetNX(values ...interface{}) *BoolCmd {
args := make([]interface{}, 1, 1+len(values))
args[0] = "msetnx"
args = appendArgs(args, pairs)
args = appendArgs(args, values)
cmd := NewBoolCmd(args...)
_ = c(cmd)
return cmd
@@ -822,7 +837,7 @@ func (c cmdable) MSetNX(pairs ...interface{}) *BoolCmd {
// Use expiration for `SETEX`-like behavior.
// Zero expiration means the key has no expiration time.
func (c cmdable) Set(key string, value interface{}, expiration time.Duration) *StatusCmd {
args := make([]interface{}, 3, 4)
args := make([]interface{}, 3, 5)
args[0] = "set"
args[1] = key
args[2] = value
@@ -954,6 +969,8 @@ func (c cmdable) HLen(key string) *IntCmd {
return cmd
}
// HMGet returns the values for the specified fields in the hash stored at key.
// It returns an interface{} to distinguish between empty string and nil value.
func (c cmdable) HMGet(key string, fields ...string) *SliceCmd {
args := make([]interface{}, 2+len(fields))
args[0] = "hmget"
@@ -966,23 +983,29 @@ func (c cmdable) HMGet(key string, fields ...string) *SliceCmd {
return cmd
}
func (c cmdable) HMSet(key string, fields map[string]interface{}) *StatusCmd {
args := make([]interface{}, 2+len(fields)*2)
args[0] = "hmset"
// HSet accepts values in following formats:
// - HMSet("myhash", "key1", "value1", "key2", "value2")
// - HMSet("myhash", []string{"key1", "value1", "key2", "value2"})
// - HMSet("myhash", map[string]interface{}{"key1": "value1", "key2": "value2"})
//
// Note that it requires Redis v4 for multiple field/value pairs support.
func (c cmdable) HSet(key string, values ...interface{}) *IntCmd {
args := make([]interface{}, 2, 2+len(values))
args[0] = "hset"
args[1] = key
i := 2
for k, v := range fields {
args[i] = k
args[i+1] = v
i += 2
}
cmd := NewStatusCmd(args...)
args = appendArgs(args, values)
cmd := NewIntCmd(args...)
_ = c(cmd)
return cmd
}
func (c cmdable) HSet(key, field string, value interface{}) *BoolCmd {
cmd := NewBoolCmd("hset", key, field, value)
// HMSet is a deprecated version of HSet left for compatibility with Redis 3.
func (c cmdable) HMSet(key string, values ...interface{}) *BoolCmd {
args := make([]interface{}, 2, 2+len(values))
args[0] = "hmset"
args[1] = key
args = appendArgs(args, values)
cmd := NewBoolCmd(args...)
_ = c(cmd)
return cmd
}
@@ -1387,7 +1410,7 @@ func (c cmdable) XRevRangeN(stream, start, stop string, count int64) *XMessageSl
}
type XReadArgs struct {
Streams []string
Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
Count int64
Block time.Duration
}
@@ -1403,6 +1426,7 @@ func (c cmdable) XRead(a *XReadArgs) *XStreamSliceCmd {
args = append(args, "block")
args = append(args, int64(a.Block/time.Millisecond))
}
args = append(args, "streams")
for _, s := range a.Streams {
args = append(args, s)
@@ -1571,6 +1595,12 @@ func (c cmdable) XTrimApprox(key string, maxLen int64) *IntCmd {
return cmd
}
func (c cmdable) XInfoGroups(key string) *XInfoGroupsCmd {
cmd := NewXInfoGroupsCmd(key)
_ = c(cmd)
return cmd
}
//------------------------------------------------------------------------------
// Z represents sorted set member.
@@ -2512,7 +2542,7 @@ func (c cmdable) GeoAdd(key string, geoLocation ...*GeoLocation) *IntCmd {
func (c cmdable) GeoRadius(key string, longitude, latitude float64, query *GeoRadiusQuery) *GeoLocationCmd {
cmd := NewGeoLocationCmd(query, "georadius_ro", key, longitude, latitude)
if query.Store != "" || query.StoreDist != "" {
cmd.setErr(errors.New("GeoRadius does not support Store or StoreDist"))
cmd.SetErr(errors.New("GeoRadius does not support Store or StoreDist"))
return cmd
}
_ = c(cmd)
@@ -2524,7 +2554,7 @@ func (c cmdable) GeoRadiusStore(key string, longitude, latitude float64, query *
args := geoLocationArgs(query, "georadius", key, longitude, latitude)
cmd := NewIntCmd(args...)
if query.Store == "" && query.StoreDist == "" {
cmd.setErr(errors.New("GeoRadiusStore requires Store or StoreDist"))
cmd.SetErr(errors.New("GeoRadiusStore requires Store or StoreDist"))
return cmd
}
_ = c(cmd)
@@ -2535,7 +2565,7 @@ func (c cmdable) GeoRadiusStore(key string, longitude, latitude float64, query *
func (c cmdable) GeoRadiusByMember(key, member string, query *GeoRadiusQuery) *GeoLocationCmd {
cmd := NewGeoLocationCmd(query, "georadiusbymember_ro", key, member)
if query.Store != "" || query.StoreDist != "" {
cmd.setErr(errors.New("GeoRadiusByMember does not support Store or StoreDist"))
cmd.SetErr(errors.New("GeoRadiusByMember does not support Store or StoreDist"))
return cmd
}
_ = c(cmd)
@@ -2547,7 +2577,7 @@ func (c cmdable) GeoRadiusByMemberStore(key, member string, query *GeoRadiusQuer
args := geoLocationArgs(query, "georadiusbymember", key, member)
cmd := NewIntCmd(args...)
if query.Store == "" && query.StoreDist == "" {
cmd.setErr(errors.New("GeoRadiusByMemberStore requires Store or StoreDist"))
cmd.SetErr(errors.New("GeoRadiusByMemberStore requires Store or StoreDist"))
return cmd
}
_ = c(cmd)
+9 -2
View File
@@ -1,8 +1,15 @@
module github.com/go-redis/redis/v7
require (
github.com/onsi/ginkgo v1.8.0
github.com/onsi/gomega v1.5.0
github.com/golang/protobuf v1.3.2 // indirect
github.com/kr/pretty v0.1.0 // indirect
github.com/onsi/ginkgo v1.10.1
github.com/onsi/gomega v1.7.0
golang.org/x/net v0.0.0-20190923162816-aa69164e4478 // indirect
golang.org/x/sys v0.0.0-20191010194322-b09406accb47 // indirect
golang.org/x/text v0.3.2 // indirect
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
gopkg.in/yaml.v2 v2.2.4 // indirect
)
go 1.11
+24 -4
View File
@@ -2,26 +2,46 @@ github.com/fsnotify/fsnotify v1.4.7 h1:IXs+QLmnXW2CcXuY+8Mzv/fWEsPGWxqefPtCP5CnV
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/hpcloud/tail v1.0.0 h1:nfCOvKYfkgYP8hkirhJocXT2+zOD8yUNjXaWfTlyFKI=
github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.8.0 h1:VkHVNpR4iVnU8XQR6DBm8BqYjN7CRzw+xKUbVVbbW9w=
github.com/onsi/ginkgo v1.8.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.5.0 h1:izbySO9zDPmjJ8rDjLvkA2zJHIo+HkYXHnf7eN7SSyo=
github.com/onsi/gomega v1.5.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/onsi/ginkgo v1.10.1 h1:q/mM8GF/n0shIN8SaAZ0V+jnLPzen6WIVZdiwrRlMlo=
github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/gomega v1.7.0 h1:XPnZz8VVBHjVsy1vzJmRwIcSwiUO+JFfrv/xGiigmME=
github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd h1:nTDtHvHSdCn1m6ITfMRqtOd/9+7a3s8RBNOZ3eYZzJA=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478 h1:l5EDrHhldLYb3ZRHDUhXF7Om7MvYXnkV9/iQNo1lX6g=
golang.org/x/net v0.0.0-20190923162816-aa69164e4478/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f h1:wMNYb4v58l5UBM7MYRLPG6ZhfOqbKu7X5eyFl8ZhKvA=
golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e h1:o3PsSEY8E4eXWkXrIP9YJALUkVZqzHJT5DOasTyn8Vs=
golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47 h1:/XfQ9z7ib8eEJX2hdgFTZJ/ntt0swNk5oYBziWeTCvY=
golang.org/x/sys v0.0.0-20191010194322-b09406accb47/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+20 -9
View File
@@ -78,7 +78,8 @@ type ConnPool struct {
stats Stats
_closed uint32 // atomic
_closed uint32 // atomic
closedCh chan struct{}
}
var _ Pooler = (*ConnPool)(nil)
@@ -90,6 +91,7 @@ func NewConnPool(opt *Options) *ConnPool {
queue: make(chan struct{}, opt.PoolSize),
conns: make([]*Conn, 0, opt.PoolSize),
idleConns: make([]*Conn, 0, opt.PoolSize),
closedCh: make(chan struct{}),
}
p.checkMinIdleConns()
@@ -416,6 +418,7 @@ func (p *ConnPool) Close() error {
if !atomic.CompareAndSwapUint32(&p._closed, 0, 1) {
return ErrClosed
}
close(p.closedCh)
var firstErr error
p.connsMu.Lock()
@@ -437,14 +440,22 @@ func (p *ConnPool) reaper(frequency time.Duration) {
ticker := time.NewTicker(frequency)
defer ticker.Stop()
for range ticker.C {
if p.closed() {
break
}
_, err := p.ReapStaleConns()
if err != nil {
internal.Logger.Printf("ReapStaleConns failed: %s", err)
continue
for {
select {
case <-ticker.C:
// It is possible that ticker and closedCh arrive together,
// and select pseudo-randomly pick ticker case, we double
// check here to prevent being executed after closed.
if p.closed() {
return
}
_, err := p.ReapStaleConns()
if err != nil {
internal.Logger.Printf("ReapStaleConns failed: %s", err)
continue
}
case <-p.closedCh:
return
}
}
}
+9 -5
View File
@@ -19,7 +19,11 @@ type BadConnError struct {
var _ error = (*BadConnError)(nil)
func (e BadConnError) Error() string {
return "pg: Conn is in a bad state"
s := "redis: Conn is in a bad state"
if e.wrapped != nil {
s += ": " + e.wrapped.Error()
}
return s
}
func (e BadConnError) Unwrap() error {
@@ -94,7 +98,7 @@ func (p *SingleConnPool) Get(ctx context.Context) (*Conn, error) {
panic("not reached")
}
}
return nil, fmt.Errorf("pg: SingleConnPool.Get: infinite loop")
return nil, fmt.Errorf("redis: SingleConnPool.Get: infinite loop")
}
func (p *SingleConnPool) Put(cn *Conn) {
@@ -166,7 +170,7 @@ func (p *SingleConnPool) Close() error {
}
}
return fmt.Errorf("pg: SingleConnPool.Close: infinite loop")
return fmt.Errorf("redis: SingleConnPool.Close: infinite loop")
}
func (p *SingleConnPool) Reset() error {
@@ -182,12 +186,12 @@ func (p *SingleConnPool) Reset() error {
p.pool.Remove(cn, ErrClosed)
p._badConnError.Store(BadConnError{wrapped: nil})
default:
return fmt.Errorf("pg: SingleConnPool does not have a Conn")
return fmt.Errorf("redis: SingleConnPool does not have a Conn")
}
if !atomic.CompareAndSwapUint32(&p.state, stateInited, stateDefault) {
state := atomic.LoadUint32(&p.state)
return fmt.Errorf("pg: invalid SingleConnPool state: %d", state)
return fmt.Errorf("redis: invalid SingleConnPool state: %d", state)
}
return nil
+20 -5
View File
@@ -21,8 +21,8 @@ type Limiter interface {
// If operation is allowed client must ReportResult of the operation
// whether it is a success or a failure.
Allow() error
// ReportResult reports the result of previously allowed operation.
// nil indicates a success, non-nil error indicates a failure.
// ReportResult reports the result of the previously allowed operation.
// nil indicates a success, non-nil error usually indicates a failure.
ReportResult(result error)
}
@@ -96,15 +96,22 @@ type Options struct {
// TLS Config to use. When set TLS will be negotiated.
TLSConfig *tls.Config
// Limiter interface used to implemented circuit breaker or rate limiter.
Limiter Limiter
}
func (opt *Options) init() {
if opt.Network == "" {
opt.Network = "tcp"
}
if opt.Addr == "" {
opt.Addr = "localhost:6379"
}
if opt.Network == "" {
if strings.HasPrefix(opt.Addr, "/") {
opt.Network = "unix"
} else {
opt.Network = "tcp"
}
}
if opt.Dialer == nil {
opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
netDialer := &net.Dialer{
@@ -145,6 +152,9 @@ func (opt *Options) init() {
opt.IdleCheckFrequency = time.Minute
}
if opt.MaxRetries == -1 {
opt.MaxRetries = 0
}
switch opt.MinRetryBackoff {
case -1:
opt.MinRetryBackoff = 0
@@ -159,6 +169,11 @@ func (opt *Options) init() {
}
}
func (opt *Options) clone() *Options {
clone := *opt
return &clone
}
// ParseURL parses an URL into Options that can be used to connect to Redis.
func ParseURL(redisURL string) (*Options, error) {
o := &Options{Network: "tcp"}
+1
View File
@@ -468,6 +468,7 @@ func (c *PubSub) initPing() {
}
c.mu.Lock()
c.reconnect(pingErr)
healthy = true
c.mu.Unlock()
}
case <-c.exit:
+89 -44
View File
@@ -32,10 +32,16 @@ type hooks struct {
hooks []Hook
}
func (hs hooks) Lock() {
func (hs *hooks) lock() {
hs.hooks = hs.hooks[:len(hs.hooks):len(hs.hooks)]
}
func (hs hooks) clone() hooks {
clone := hs
clone.lock()
return clone
}
func (hs *hooks) AddHook(hook Hook) {
hs.hooks = append(hs.hooks, hook)
}
@@ -45,13 +51,14 @@ func (hs hooks) process(
) error {
ctx, err := hs.beforeProcess(ctx, cmd)
if err != nil {
cmd.SetErr(err)
return err
}
cmdErr := fn(ctx, cmd)
err = hs.afterProcess(ctx, cmd)
if err != nil {
if err := hs.afterProcess(ctx, cmd); err != nil {
cmd.SetErr(err)
return err
}
@@ -85,13 +92,14 @@ func (hs hooks) processPipeline(
) error {
ctx, err := hs.beforeProcessPipeline(ctx, cmds)
if err != nil {
setCmdsErr(cmds, err)
return err
}
cmdsErr := fn(ctx, cmds)
err = hs.afterProcessPipeline(ctx, cmds)
if err != nil {
if err := hs.afterProcessPipeline(ctx, cmds); err != nil {
setCmdsErr(cmds, err)
return err
}
@@ -120,16 +128,45 @@ func (hs hooks) afterProcessPipeline(ctx context.Context, cmds []Cmder) error {
return firstErr
}
func (hs hooks) processTxPipeline(
ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error,
) error {
cmds = wrapMultiExec(cmds)
return hs.processPipeline(ctx, cmds, fn)
}
//------------------------------------------------------------------------------
type baseClient struct {
opt *Options
connPool pool.Pooler
limiter Limiter
onClose func() error // hook called when client is closed
}
func newBaseClient(opt *Options, connPool pool.Pooler) *baseClient {
return &baseClient{
opt: opt,
connPool: connPool,
}
}
func (c *baseClient) clone() *baseClient {
clone := *c
return &clone
}
func (c *baseClient) withTimeout(timeout time.Duration) *baseClient {
opt := c.opt.clone()
opt.ReadTimeout = timeout
opt.WriteTimeout = timeout
clone := c.clone()
clone.opt = opt
return clone
}
func (c *baseClient) String() string {
return fmt.Sprintf("Redis<%s db:%d>", c.getAddr(), c.opt.DB)
}
@@ -150,8 +187,8 @@ func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error) {
}
func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error) {
if c.limiter != nil {
err := c.limiter.Allow()
if c.opt.Limiter != nil {
err := c.opt.Limiter.Allow()
if err != nil {
return nil, err
}
@@ -159,8 +196,8 @@ func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error) {
cn, err := c._getConn(ctx)
if err != nil {
if c.limiter != nil {
c.limiter.ReportResult(err)
if c.opt.Limiter != nil {
c.opt.Limiter.ReportResult(err)
}
return nil, err
}
@@ -228,8 +265,8 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
}
func (c *baseClient) releaseConn(cn *pool.Conn, err error) {
if c.limiter != nil {
c.limiter.ReportResult(err)
if c.opt.Limiter != nil {
c.opt.Limiter.ReportResult(err)
}
if isBadConn(err, false) {
@@ -257,7 +294,7 @@ func (c *baseClient) withConn(
func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
err := c._process(ctx, cmd)
if err != nil {
cmd.setErr(err)
cmd.SetErr(err)
return err
}
return nil
@@ -381,7 +418,7 @@ func (c *baseClient) pipelineProcessCmds(
ctx context.Context, cn *pool.Conn, cmds []Cmder,
) (bool, error) {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmd(wr, cmds...)
return writeCmds(wr, cmds)
})
if err != nil {
return true, err
@@ -407,41 +444,46 @@ func (c *baseClient) txPipelineProcessCmds(
ctx context.Context, cn *pool.Conn, cmds []Cmder,
) (bool, error) {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return txPipelineWriteMulti(wr, cmds)
return writeCmds(wr, cmds)
})
if err != nil {
return true, err
}
err = cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
err := txPipelineReadQueued(rd, cmds)
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
cmds = cmds[1 : len(cmds)-1]
err := txPipelineReadQueued(rd, statusCmd, cmds)
if err != nil {
return err
}
return pipelineReadCmds(rd, cmds)
})
return false, err
}
func txPipelineWriteMulti(wr *proto.Writer, cmds []Cmder) error {
multiExec := make([]Cmder, 0, len(cmds)+2)
multiExec = append(multiExec, NewStatusCmd("MULTI"))
multiExec = append(multiExec, cmds...)
multiExec = append(multiExec, NewSliceCmd("EXEC"))
return writeCmd(wr, multiExec...)
func wrapMultiExec(cmds []Cmder) []Cmder {
if len(cmds) == 0 {
panic("not reached")
}
cmds = append(cmds, make([]Cmder, 2)...)
copy(cmds[1:], cmds[:len(cmds)-2])
cmds[0] = NewStatusCmd("multi")
cmds[len(cmds)-1] = NewSliceCmd("exec")
return cmds
}
func txPipelineReadQueued(rd *proto.Reader, cmds []Cmder) error {
func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder) error {
// Parse queued replies.
var statusCmd StatusCmd
err := statusCmd.readReply(rd)
if err != nil {
if err := statusCmd.readReply(rd); err != nil {
return err
}
for range cmds {
err = statusCmd.readReply(rd)
if err != nil && !isRedisError(err) {
if err := statusCmd.readReply(rd); err != nil && !isRedisError(err) {
return err
}
}
@@ -474,7 +516,7 @@ func txPipelineReadQueued(rd *proto.Reader, cmds []Cmder) error {
// underlying connections. It's safe for concurrent use by multiple
// goroutines.
type Client struct {
baseClient
*baseClient
cmdable
hooks
ctx context.Context
@@ -485,17 +527,27 @@ func NewClient(opt *Options) *Client {
opt.init()
c := Client{
baseClient: baseClient{
opt: opt,
connPool: newConnPool(opt),
},
ctx: context.Background(),
baseClient: newBaseClient(opt, newConnPool(opt)),
ctx: context.Background(),
}
c.cmdable = c.Process
return &c
}
func (c *Client) clone() *Client {
clone := *c
clone.cmdable = clone.Process
clone.hooks.lock()
return &clone
}
func (c *Client) WithTimeout(timeout time.Duration) *Client {
clone := c.clone()
clone.baseClient = c.baseClient.withTimeout(timeout)
return clone
}
func (c *Client) Context() context.Context {
return c.ctx
}
@@ -504,11 +556,9 @@ func (c *Client) WithContext(ctx context.Context) *Client {
if ctx == nil {
panic("nil context")
}
clone := *c
clone.cmdable = clone.Process
clone.hooks.Lock()
clone := c.clone()
clone.ctx = ctx
return &clone
return clone
}
func (c *Client) Conn() *Conn {
@@ -539,7 +589,7 @@ func (c *Client) processPipeline(ctx context.Context, cmds []Cmder) error {
}
func (c *Client) processTxPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, c.baseClient.processTxPipeline)
return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline)
}
// Options returns read-only Options that were used to create the client.
@@ -547,11 +597,6 @@ func (c *Client) Options() *Options {
return c.opt
}
func (c *Client) SetLimiter(l Limiter) *Client {
c.limiter = l
return c
}
type PoolStats pool.Stats
// PoolStats returns connection pool stats.
+42 -18
View File
@@ -6,7 +6,7 @@ import "time"
func NewCmdResult(val interface{}, err error) *Cmd {
var cmd Cmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -14,7 +14,7 @@ func NewCmdResult(val interface{}, err error) *Cmd {
func NewSliceResult(val []interface{}, err error) *SliceCmd {
var cmd SliceCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -22,7 +22,7 @@ func NewSliceResult(val []interface{}, err error) *SliceCmd {
func NewStatusResult(val string, err error) *StatusCmd {
var cmd StatusCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -30,7 +30,7 @@ func NewStatusResult(val string, err error) *StatusCmd {
func NewIntResult(val int64, err error) *IntCmd {
var cmd IntCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -38,7 +38,7 @@ func NewIntResult(val int64, err error) *IntCmd {
func NewDurationResult(val time.Duration, err error) *DurationCmd {
var cmd DurationCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -46,7 +46,7 @@ func NewDurationResult(val time.Duration, err error) *DurationCmd {
func NewBoolResult(val bool, err error) *BoolCmd {
var cmd BoolCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -54,7 +54,7 @@ func NewBoolResult(val bool, err error) *BoolCmd {
func NewStringResult(val string, err error) *StringCmd {
var cmd StringCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -62,7 +62,7 @@ func NewStringResult(val string, err error) *StringCmd {
func NewFloatResult(val float64, err error) *FloatCmd {
var cmd FloatCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -70,7 +70,7 @@ func NewFloatResult(val float64, err error) *FloatCmd {
func NewStringSliceResult(val []string, err error) *StringSliceCmd {
var cmd StringSliceCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -78,7 +78,7 @@ func NewStringSliceResult(val []string, err error) *StringSliceCmd {
func NewBoolSliceResult(val []bool, err error) *BoolSliceCmd {
var cmd BoolSliceCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -86,7 +86,7 @@ func NewBoolSliceResult(val []bool, err error) *BoolSliceCmd {
func NewStringStringMapResult(val map[string]string, err error) *StringStringMapCmd {
var cmd StringStringMapCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -94,7 +94,7 @@ func NewStringStringMapResult(val map[string]string, err error) *StringStringMap
func NewStringIntMapCmdResult(val map[string]int64, err error) *StringIntMapCmd {
var cmd StringIntMapCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -102,7 +102,7 @@ func NewStringIntMapCmdResult(val map[string]int64, err error) *StringIntMapCmd
func NewZSliceCmdResult(val []Z, err error) *ZSliceCmd {
var cmd ZSliceCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -110,7 +110,7 @@ func NewZSliceCmdResult(val []Z, err error) *ZSliceCmd {
func NewZWithKeyCmdResult(val *ZWithKey, err error) *ZWithKeyCmd {
var cmd ZWithKeyCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -119,7 +119,7 @@ func NewScanCmdResult(keys []string, cursor uint64, err error) *ScanCmd {
var cmd ScanCmd
cmd.page = keys
cmd.cursor = cursor
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -127,7 +127,7 @@ func NewScanCmdResult(keys []string, cursor uint64, err error) *ScanCmd {
func NewClusterSlotsCmdResult(val []ClusterSlot, err error) *ClusterSlotsCmd {
var cmd ClusterSlotsCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
@@ -135,7 +135,15 @@ func NewClusterSlotsCmdResult(val []ClusterSlot, err error) *ClusterSlotsCmd {
func NewGeoLocationCmdResult(val []GeoLocation, err error) *GeoLocationCmd {
var cmd GeoLocationCmd
cmd.locations = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
// NewGeoPosCmdResult returns a GeoPosCmd initialised with val and err for testing
func NewGeoPosCmdResult(val []*GeoPos, err error) *GeoPosCmd {
var cmd GeoPosCmd
cmd.val = val
cmd.SetErr(err)
return &cmd
}
@@ -143,6 +151,22 @@ func NewGeoLocationCmdResult(val []GeoLocation, err error) *GeoLocationCmd {
func NewCommandsInfoCmdResult(val map[string]*CommandInfo, err error) *CommandsInfoCmd {
var cmd CommandsInfoCmd
cmd.val = val
cmd.setErr(err)
cmd.SetErr(err)
return &cmd
}
// NewXMessageSliceCmdResult returns a XMessageSliceCmd initialised with val and err for testing
func NewXMessageSliceCmdResult(val []XMessage, err error) *XMessageSliceCmd {
var cmd XMessageSliceCmd
cmd.val = val
cmd.SetErr(err)
return &cmd
}
// NewXStreamSliceCmdResult returns a XStreamSliceCmd initialised with val and err for testing
func NewXStreamSliceCmdResult(val []XStream, err error) *XStreamSliceCmd {
var cmd XStreamSliceCmd
cmd.val = val
cmd.SetErr(err)
return &cmd
}
+7 -11
View File
@@ -407,7 +407,7 @@ func (c *Ring) WithContext(ctx context.Context) *Ring {
}
clone := *c
clone.cmdable = clone.Process
clone.hooks.Lock()
clone.hooks.lock()
clone.ctx = ctx
return &clone
}
@@ -561,7 +561,7 @@ func (c *Ring) cmdShard(cmd Cmder) (*ringShard, error) {
func (c *Ring) process(ctx context.Context, cmd Cmder) error {
err := c._process(ctx, cmd)
if err != nil {
cmd.setErr(err)
cmd.SetErr(err)
return err
}
return nil
@@ -581,7 +581,7 @@ func (c *Ring) _process(ctx context.Context, cmd Cmder) error {
return err
}
lastErr = shard.Client._process(ctx, cmd)
lastErr = shard.Client.ProcessContext(ctx, cmd)
if lastErr == nil || !isRetryableError(lastErr, cmd.readTimeout() == nil) {
return lastErr
}
@@ -646,10 +646,7 @@ func (c *Ring) generalProcessPipeline(
go func(hash string, cmds []Cmder) {
defer wg.Done()
err := c.processShardPipeline(ctx, hash, cmds, tx)
if err != nil {
setCmdsErr(cmds, err)
}
_ = c.processShardPipeline(ctx, hash, cmds, tx)
}(hash, cmds)
}
@@ -663,15 +660,14 @@ func (c *Ring) processShardPipeline(
//TODO: retry?
shard, err := c.shards.GetByHash(hash)
if err != nil {
setCmdsErr(cmds, err)
return err
}
if tx {
err = shard.Client._generalProcessPipeline(
ctx, cmds, shard.Client.txPipelineProcessCmds)
err = shard.Client.processTxPipeline(ctx, cmds)
} else {
err = shard.Client._generalProcessPipeline(
ctx, cmds, shard.Client.pipelineProcessCmds)
err = shard.Client.processPipeline(ctx, cmds)
}
return err
}
+51 -42
View File
@@ -59,7 +59,9 @@ func (opt *FailoverOptions) options() *Options {
DB: opt.DB,
Password: opt.Password,
MaxRetries: opt.MaxRetries,
MaxRetries: opt.MaxRetries,
MinRetryBackoff: opt.MinRetryBackoff,
MaxRetryBackoff: opt.MaxRetryBackoff,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
@@ -69,6 +71,8 @@ func (opt *FailoverOptions) options() *Options {
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: opt.IdleCheckFrequency,
MinIdleConns: opt.MinIdleConns,
MaxConnAge: opt.MaxConnAge,
TLSConfig: opt.TLSConfig,
}
@@ -90,14 +94,11 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
}
c := Client{
baseClient: baseClient{
opt: opt,
connPool: failover.Pool(),
onClose: failover.Close,
},
ctx: context.Background(),
baseClient: newBaseClient(opt, failover.Pool()),
ctx: context.Background(),
}
c.cmdable = c.Process
c.onClose = failover.Close
return &c
}
@@ -301,6 +302,19 @@ func (c *sentinelFailover) Close() error {
return nil
}
func (c *sentinelFailover) closeSentinel() error {
firstErr := c.pubsub.Close()
c.pubsub = nil
err := c.sentinel.Close()
if err != nil && firstErr == nil {
firstErr = err
}
c.sentinel = nil
return firstErr
}
func (c *sentinelFailover) Pool() *pool.ConnPool {
c.poolOnce.Do(func() {
opt := *c.opt
@@ -331,17 +345,32 @@ func (c *sentinelFailover) MasterAddr() (string, error) {
}
func (c *sentinelFailover) masterAddr() (string, error) {
addr := c.getMasterAddr()
if addr != "" {
return addr, nil
c.mu.RLock()
sentinel := c.sentinel
c.mu.RUnlock()
if sentinel != nil {
addr := c.getMasterAddr(sentinel)
if addr != "" {
return addr, nil
}
}
c.mu.Lock()
defer c.mu.Unlock()
if c.sentinel != nil {
addr := c.getMasterAddr(c.sentinel)
if addr != "" {
return addr, nil
}
_ = c.closeSentinel()
}
for i, sentinelAddr := range c.sentinelAddrs {
sentinel := NewSentinelClient(&Options{
Addr: sentinelAddr,
Addr: sentinelAddr,
Dialer: c.opt.Dialer,
Password: c.password,
@@ -378,27 +407,13 @@ func (c *sentinelFailover) masterAddr() (string, error) {
return "", errors.New("redis: all sentinels are unreachable")
}
func (c *sentinelFailover) getMasterAddr() string {
c.mu.RLock()
sentinel := c.sentinel
c.mu.RUnlock()
if sentinel == nil {
return ""
}
func (c *sentinelFailover) getMasterAddr(sentinel *SentinelClient) string {
addr, err := sentinel.GetMasterAddrByName(c.masterName).Result()
if err != nil {
internal.Logger.Printf("sentinel: GetMasterAddrByName name=%q failed: %s",
c.masterName, err)
c.mu.Lock()
if c.sentinel == sentinel {
_ = c.closeSentinel()
}
c.mu.Unlock()
return ""
}
return net.JoinHostPort(addr[0], addr[1])
}
@@ -413,6 +428,10 @@ func (c *sentinelFailover) switchMaster(addr string) {
c.mu.Lock()
defer c.mu.Unlock()
if c._masterAddr == addr {
return
}
internal.Logger.Printf("sentinel: new master=%q addr=%q",
c.masterName, addr)
_ = c.Pool().Filter(func(cn *pool.Conn) bool {
@@ -422,28 +441,18 @@ func (c *sentinelFailover) switchMaster(addr string) {
}
func (c *sentinelFailover) setSentinel(sentinel *SentinelClient) {
c.discoverSentinels(sentinel)
if c.sentinel != nil {
panic("not reached")
}
c.sentinel = sentinel
c.discoverSentinels()
c.pubsub = sentinel.Subscribe("+switch-master")
go c.listen(c.pubsub)
}
func (c *sentinelFailover) closeSentinel() error {
firstErr := c.pubsub.Close()
c.pubsub = nil
err := c.sentinel.Close()
if err != nil && firstErr == nil {
firstErr = err
}
c.sentinel = nil
return firstErr
}
func (c *sentinelFailover) discoverSentinels(sentinel *SentinelClient) {
sentinels, err := sentinel.Sentinels(c.masterName).Result()
func (c *sentinelFailover) discoverSentinels() {
sentinels, err := c.sentinel.Sentinels(c.masterName).Result()
if err != nil {
internal.Logger.Printf("sentinel: Sentinels master=%q failed: %s", c.masterName, err)
return
+28 -15
View File
@@ -18,6 +18,7 @@ type Tx struct {
baseClient
cmdable
statefulCmdable
hooks
ctx context.Context
}
@@ -27,7 +28,8 @@ func (c *Client) newTx(ctx context.Context) *Tx {
opt: c.opt,
connPool: pool.NewStickyConnPool(c.connPool.(*pool.ConnPool), true),
},
ctx: ctx,
hooks: c.hooks.clone(),
ctx: ctx,
}
tx.init()
return &tx
@@ -47,8 +49,9 @@ func (c *Tx) WithContext(ctx context.Context) *Tx {
panic("nil context")
}
clone := *c
clone.ctx = ctx
clone.init()
clone.hooks.lock()
clone.ctx = ctx
return &clone
}
@@ -57,7 +60,7 @@ func (c *Tx) Process(cmd Cmder) error {
}
func (c *Tx) ProcessContext(ctx context.Context, cmd Cmder) error {
return c.baseClient.process(ctx, cmd)
return c.hooks.process(ctx, cmd, c.baseClient.process)
}
// Watch prepares a transaction and marks the keys to be watched
@@ -113,17 +116,25 @@ func (c *Tx) Unwatch(keys ...string) *StatusCmd {
return cmd
}
// Pipeline creates a new pipeline. It is more convenient to use Pipelined.
// Pipeline creates a pipeline. Usually it is more convenient to use Pipelined.
func (c *Tx) Pipeline() Pipeliner {
pipe := Pipeline{
ctx: c.ctx,
exec: c.processTxPipeline,
ctx: c.ctx,
exec: func(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, c.baseClient.processPipeline)
},
}
pipe.init()
return &pipe
}
// Pipelined executes commands queued in the fn in a transaction.
// Pipelined executes commands queued in the fn outside of the transaction.
// Use TxPipelined if you need transactional behavior.
func (c *Tx) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(fn)
}
// TxPipelined executes commands queued in the fn in the transaction.
//
// When using WATCH, EXEC will execute commands only if the watched keys
// were not modified, allowing for a check-and-set mechanism.
@@ -131,16 +142,18 @@ func (c *Tx) Pipeline() Pipeliner {
// Exec always returns list of commands. If transaction fails
// TxFailedErr is returned. Otherwise Exec returns an error of the first
// failed command or nil.
func (c *Tx) Pipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(fn)
}
// TxPipelined is an alias for Pipelined.
func (c *Tx) TxPipelined(fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipelined(fn)
return c.TxPipeline().Pipelined(fn)
}
// TxPipeline is an alias for Pipeline.
// TxPipeline creates a pipeline. Usually it is more convenient to use TxPipelined.
func (c *Tx) TxPipeline() Pipeliner {
return c.Pipeline()
pipe := Pipeline{
ctx: c.ctx,
exec: func(ctx context.Context, cmds []Cmder) error {
return c.hooks.processTxPipeline(ctx, cmds, c.baseClient.processTxPipeline)
},
}
pipe.init()
return &pipe
}
+9 -6
View File
@@ -49,7 +49,8 @@ type UniversalOptions struct {
MasterName string
}
func (o *UniversalOptions) cluster() *ClusterOptions {
// Cluster returns cluster options created from the universal options.
func (o *UniversalOptions) Cluster() *ClusterOptions {
if len(o.Addrs) == 0 {
o.Addrs = []string{"127.0.0.1:6379"}
}
@@ -84,7 +85,8 @@ func (o *UniversalOptions) cluster() *ClusterOptions {
}
}
func (o *UniversalOptions) failover() *FailoverOptions {
// Failover returns failover options created from the universal options.
func (o *UniversalOptions) Failover() *FailoverOptions {
if len(o.Addrs) == 0 {
o.Addrs = []string{"127.0.0.1:26379"}
}
@@ -118,7 +120,8 @@ func (o *UniversalOptions) failover() *FailoverOptions {
}
}
func (o *UniversalOptions) simple() *Options {
// Simple returns basic options created from the universal options.
func (o *UniversalOptions) Simple() *Options {
addr := "127.0.0.1:6379"
if len(o.Addrs) > 0 {
addr = o.Addrs[0]
@@ -183,9 +186,9 @@ var _ UniversalClient = (*Ring)(nil)
// 3. otherwise, a single-node redis Client will be returned.
func NewUniversalClient(opts *UniversalOptions) UniversalClient {
if opts.MasterName != "" {
return NewFailoverClient(opts.failover())
return NewFailoverClient(opts.Failover())
} else if len(opts.Addrs) > 1 {
return NewClusterClient(opts.cluster())
return NewClusterClient(opts.Cluster())
}
return NewClient(opts.simple())
return NewClient(opts.Simple())
}
+19 -3
View File
@@ -37,7 +37,7 @@ func (c *Cache) Purge() {
c.lock.Unlock()
}
// Add adds a value to the cache. Returns true if an eviction occurred.
// Add adds a value to the cache. Returns true if an eviction occurred.
func (c *Cache) Add(key, value interface{}) (evicted bool) {
c.lock.Lock()
evicted = c.lru.Add(key, value)
@@ -71,8 +71,8 @@ func (c *Cache) Peek(key interface{}) (value interface{}, ok bool) {
return value, ok
}
// ContainsOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// ContainsOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// Returns whether found and whether an eviction occurred.
func (c *Cache) ContainsOrAdd(key, value interface{}) (ok, evicted bool) {
c.lock.Lock()
@@ -85,6 +85,22 @@ func (c *Cache) ContainsOrAdd(key, value interface{}) (ok, evicted bool) {
return false, evicted
}
// PeekOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// Returns whether found and whether an eviction occurred.
func (c *Cache) PeekOrAdd(key, value interface{}) (previous interface{}, ok, evicted bool) {
c.lock.Lock()
defer c.lock.Unlock()
previous, ok = c.lru.Peek(key)
if ok {
return previous, true, false
}
evicted = c.lru.Add(key, value)
return nil, false, evicted
}
// Remove removes the provided key from the cache.
func (c *Cache) Remove(key interface{}) (present bool) {
c.lock.Lock()
+38
View File
@@ -229,6 +229,10 @@ type decoder struct {
mapType reflect.Type
terrors []string
strict bool
decodeCount int
aliasCount int
aliasDepth int
}
var (
@@ -314,7 +318,39 @@ func (d *decoder) prepare(n *node, out reflect.Value) (newout reflect.Value, unm
return out, false, false
}
const (
// 400,000 decode operations is ~500kb of dense object declarations, or ~5kb of dense object declarations with 10000% alias expansion
alias_ratio_range_low = 400000
// 4,000,000 decode operations is ~5MB of dense object declarations, or ~4.5MB of dense object declarations with 10% alias expansion
alias_ratio_range_high = 4000000
// alias_ratio_range is the range over which we scale allowed alias ratios
alias_ratio_range = float64(alias_ratio_range_high - alias_ratio_range_low)
)
func allowedAliasRatio(decodeCount int) float64 {
switch {
case decodeCount <= alias_ratio_range_low:
// allow 99% to come from alias expansion for small-to-medium documents
return 0.99
case decodeCount >= alias_ratio_range_high:
// allow 10% to come from alias expansion for very large documents
return 0.10
default:
// scale smoothly from 99% down to 10% over the range.
// this maps to 396,000 - 400,000 allowed alias-driven decodes over the range.
// 400,000 decode operations is ~100MB of allocations in worst-case scenarios (single-item maps).
return 0.99 - 0.89*(float64(decodeCount-alias_ratio_range_low)/alias_ratio_range)
}
}
func (d *decoder) unmarshal(n *node, out reflect.Value) (good bool) {
d.decodeCount++
if d.aliasDepth > 0 {
d.aliasCount++
}
if d.aliasCount > 100 && d.decodeCount > 1000 && float64(d.aliasCount)/float64(d.decodeCount) > allowedAliasRatio(d.decodeCount) {
failf("document contains excessive aliasing")
}
switch n.kind {
case documentNode:
return d.document(n, out)
@@ -353,7 +389,9 @@ func (d *decoder) alias(n *node, out reflect.Value) (good bool) {
failf("anchor '%s' value contains itself", n.value)
}
d.aliases[n] = true
d.aliasDepth++
good = d.unmarshal(n.alias, out)
d.aliasDepth--
delete(d.aliases, n)
return good
}
+1 -1
View File
@@ -81,7 +81,7 @@ func resolvableTag(tag string) bool {
return false
}
var yamlStyleFloat = regexp.MustCompile(`^[-+]?[0-9]*\.?[0-9]+([eE][-+][0-9]+)?$`)
var yamlStyleFloat = regexp.MustCompile(`^[-+]?(\.[0-9]+|[0-9]+(\.[0-9]*)?)([eE][-+]?[0-9]+)?$`)
func resolve(tag string, in string) (rtag string, out interface{}) {
if !resolvableTag(tag) {
+16
View File
@@ -906,6 +906,9 @@ func yaml_parser_remove_simple_key(parser *yaml_parser_t) bool {
return true
}
// max_flow_level limits the flow_level
const max_flow_level = 10000
// Increase the flow level and resize the simple key list if needed.
func yaml_parser_increase_flow_level(parser *yaml_parser_t) bool {
// Reset the simple key on the next level.
@@ -913,6 +916,11 @@ func yaml_parser_increase_flow_level(parser *yaml_parser_t) bool {
// Increase the flow level.
parser.flow_level++
if parser.flow_level > max_flow_level {
return yaml_parser_set_scanner_error(parser,
"while increasing flow level", parser.simple_keys[len(parser.simple_keys)-1].mark,
fmt.Sprintf("exceeded max depth of %d", max_flow_level))
}
return true
}
@@ -925,6 +933,9 @@ func yaml_parser_decrease_flow_level(parser *yaml_parser_t) bool {
return true
}
// max_indents limits the indents stack size
const max_indents = 10000
// Push the current indentation level to the stack and set the new level
// the current column is greater than the indentation level. In this case,
// append or insert the specified token into the token queue.
@@ -939,6 +950,11 @@ func yaml_parser_roll_indent(parser *yaml_parser_t, column, number int, typ yaml
// indentation level.
parser.indents = append(parser.indents, parser.indent)
parser.indent = column
if len(parser.indents) > max_indents {
return yaml_parser_set_scanner_error(parser,
"while increasing indent level", parser.simple_keys[len(parser.simple_keys)-1].mark,
fmt.Sprintf("exceeded max depth of %d", max_indents))
}
// Create a token and insert it into the queue.
token := yaml_token_t{
+5 -4
View File
@@ -84,9 +84,10 @@ github.com/go-pkgz/auth/token
# github.com/go-pkgz/jrpc v0.1.0
## explicit
github.com/go-pkgz/jrpc
# github.com/go-pkgz/lcw v0.5.0
# github.com/go-pkgz/lcw v0.5.1-0.20200509170726-dc283cfc28cf
## explicit
github.com/go-pkgz/lcw
github.com/go-pkgz/lcw/internal/cache
# github.com/go-pkgz/lgr v0.7.0
## explicit
github.com/go-pkgz/lgr
@@ -101,7 +102,7 @@ github.com/go-pkgz/rest/logger
# github.com/go-pkgz/syncs v1.1.1
## explicit
github.com/go-pkgz/syncs
# github.com/go-redis/redis/v7 v7.0.0-beta.4
# github.com/go-redis/redis/v7 v7.2.0
github.com/go-redis/redis/v7
github.com/go-redis/redis/v7/internal
github.com/go-redis/redis/v7/internal/consistenthash
@@ -126,7 +127,7 @@ github.com/hashicorp/errwrap
# github.com/hashicorp/go-multierror v1.1.0
## explicit
github.com/hashicorp/go-multierror
# github.com/hashicorp/golang-lru v0.5.3
# github.com/hashicorp/golang-lru v0.5.4
github.com/hashicorp/golang-lru
github.com/hashicorp/golang-lru/simplelru
# github.com/kyokomi/emoji v2.2.1+incompatible
@@ -254,5 +255,5 @@ google.golang.org/appengine/urlfetch
gopkg.in/oauth2.v3
gopkg.in/oauth2.v3/errors
gopkg.in/oauth2.v3/server
# gopkg.in/yaml.v2 v2.2.2
# gopkg.in/yaml.v2 v2.2.4
gopkg.in/yaml.v2