actix/actix-extras
1
0
Fork 0
mirror of https://github.com/actix/actix-extras.git synced 2025-03-12 08:42:47 +01:00
Code Issues Releases Wiki Activity
actix-extras/src/actix_redis/redis.rs.html

290 lines
24 KiB
HTML
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

<!DOCTYPE html><html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><meta name="generator" content="rustdoc"><meta name="description" content="Source of the Rust file `actix-redis/src/redis.rs`."><meta name="keywords" content="rust, rustlang, rust-lang"><title>redis.rs - source</title><link rel="preload" as="font" type="font/woff2" crossorigin href="../../SourceSerif4-Regular.ttf.woff2"><link rel="preload" as="font" type="font/woff2" crossorigin href="../../FiraSans-Regular.woff2"><link rel="preload" as="font" type="font/woff2" crossorigin href="../../FiraSans-Medium.woff2"><link rel="preload" as="font" type="font/woff2" crossorigin href="../../SourceCodePro-Regular.ttf.woff2"><link rel="preload" as="font" type="font/woff2" crossorigin href="../../SourceSerif4-Bold.ttf.woff2"><link rel="preload" as="font" type="font/woff2" crossorigin href="../../SourceCodePro-Semibold.ttf.woff2"><link rel="stylesheet" type="text/css" href="../../normalize.css"><link rel="stylesheet" type="text/css" href="../../rustdoc.css" id="mainThemeStyle"><link rel="stylesheet" type="text/css" href="../../ayu.css" disabled><link rel="stylesheet" type="text/css" href="../../dark.css" disabled><link rel="stylesheet" type="text/css" href="../../light.css" id="themeStyle"><script id="default-settings" ></script><script src="../../storage.js"></script><script src="../../crates.js"></script><script defer src="../../main.js"></script><script defer src="../../source-script.js"></script><script defer src="../../source-files.js"></script>
<noscript><link rel="stylesheet" href="../../noscript.css"></noscript><link rel="alternate icon" type="image/png" href="../../favicon-16x16.png"><link rel="alternate icon" type="image/png" href="../../favicon-32x32.png"><link rel="icon" type="image/svg+xml" href="../../favicon.svg"></head><body class="rustdoc source"><!--[if lte IE 11]><div class="warning">This old browser is unsupported and will most likely display funky things.</div><![endif]--><nav class="mobile-topbar"><button class="sidebar-menu-toggle">&#9776;</button><a class="sidebar-logo" href="../../actix_redis/index.html"><div class="logo-container"><img class="rust-logo" src="../../rust-logo.svg" alt="logo"></div>
</a><h2 class="location"></h2>
</nav>
<nav class="sidebar"><a class="sidebar-logo" href="../../actix_redis/index.html"><div class="logo-container"><img class="rust-logo" src="../../rust-logo.svg" alt="logo"></div>
</a></nav><main><div class="width-limiter"><div class="sub-container"><a class="sub-logo-container" href="../../actix_redis/index.html"><img class="rust-logo" src="../../rust-logo.svg" alt="logo"></a><nav class="sub"><div class="theme-picker hidden"><button id="theme-picker" aria-label="Pick another theme!" aria-haspopup="menu" title="themes"><img width="22" height="22" alt="Pick another theme!" src="../../brush.svg"></button><div id="theme-choices" role="menu"></div></div><form class="search-form"><div class="search-container"><span></span><input class="search-input" name="search" autocomplete="off" spellcheck="false" placeholder="Click or press ‘S’ to search, ‘?’ for more options…" type="search"><button type="button" id="help-button" title="help">?</button><a id="settings-menu" href="../../settings.html" title="settings"><img width="22" height="22" alt="Change settings" src="../../wheel.svg"></a></div></form></nav></div><section id="main-content" class="content"><div class="example-wrap"><pre class="line-numbers"><span id="1"> 1</span>
<span id="2"> 2</span>
<span id="3"> 3</span>
<span id="4"> 4</span>
<span id="5"> 5</span>
<span id="6"> 6</span>
<span id="7"> 7</span>
<span id="8"> 8</span>
<span id="9"> 9</span>
<span id="10"> 10</span>
<span id="11"> 11</span>
<span id="12"> 12</span>
<span id="13"> 13</span>
<span id="14"> 14</span>
<span id="15"> 15</span>
<span id="16"> 16</span>
<span id="17"> 17</span>
<span id="18"> 18</span>
<span id="19"> 19</span>
<span id="20"> 20</span>
<span id="21"> 21</span>
<span id="22"> 22</span>
<span id="23"> 23</span>
<span id="24"> 24</span>
<span id="25"> 25</span>
<span id="26"> 26</span>
<span id="27"> 27</span>
<span id="28"> 28</span>
<span id="29"> 29</span>
<span id="30"> 30</span>
<span id="31"> 31</span>
<span id="32"> 32</span>
<span id="33"> 33</span>
<span id="34"> 34</span>
<span id="35"> 35</span>
<span id="36"> 36</span>
<span id="37"> 37</span>
<span id="38"> 38</span>
<span id="39"> 39</span>
<span id="40"> 40</span>
<span id="41"> 41</span>
<span id="42"> 42</span>
<span id="43"> 43</span>
<span id="44"> 44</span>
<span id="45"> 45</span>
<span id="46"> 46</span>
<span id="47"> 47</span>
<span id="48"> 48</span>
<span id="49"> 49</span>
<span id="50"> 50</span>
<span id="51"> 51</span>
<span id="52"> 52</span>
<span id="53"> 53</span>
<span id="54"> 54</span>
<span id="55"> 55</span>
<span id="56"> 56</span>
<span id="57"> 57</span>
<span id="58"> 58</span>
<span id="59"> 59</span>
<span id="60"> 60</span>
<span id="61"> 61</span>
<span id="62"> 62</span>
<span id="63"> 63</span>
<span id="64"> 64</span>
<span id="65"> 65</span>
<span id="66"> 66</span>
<span id="67"> 67</span>
<span id="68"> 68</span>
<span id="69"> 69</span>
<span id="70"> 70</span>
<span id="71"> 71</span>
<span id="72"> 72</span>
<span id="73"> 73</span>
<span id="74"> 74</span>
<span id="75"> 75</span>
<span id="76"> 76</span>
<span id="77"> 77</span>
<span id="78"> 78</span>
<span id="79"> 79</span>
<span id="80"> 80</span>
<span id="81"> 81</span>
<span id="82"> 82</span>
<span id="83"> 83</span>
<span id="84"> 84</span>
<span id="85"> 85</span>
<span id="86"> 86</span>
<span id="87"> 87</span>
<span id="88"> 88</span>
<span id="89"> 89</span>
<span id="90"> 90</span>
<span id="91"> 91</span>
<span id="92"> 92</span>
<span id="93"> 93</span>
<span id="94"> 94</span>
<span id="95"> 95</span>
<span id="96"> 96</span>
<span id="97"> 97</span>
<span id="98"> 98</span>
<span id="99"> 99</span>
<span id="100">100</span>
<span id="101">101</span>
<span id="102">102</span>
<span id="103">103</span>
<span id="104">104</span>
<span id="105">105</span>
<span id="106">106</span>
<span id="107">107</span>
<span id="108">108</span>
<span id="109">109</span>
<span id="110">110</span>
<span id="111">111</span>
<span id="112">112</span>
<span id="113">113</span>
<span id="114">114</span>
<span id="115">115</span>
<span id="116">116</span>
<span id="117">117</span>
<span id="118">118</span>
<span id="119">119</span>
<span id="120">120</span>
<span id="121">121</span>
<span id="122">122</span>
<span id="123">123</span>
<span id="124">124</span>
<span id="125">125</span>
<span id="126">126</span>
<span id="127">127</span>
<span id="128">128</span>
<span id="129">129</span>
<span id="130">130</span>
<span id="131">131</span>
<span id="132">132</span>
<span id="133">133</span>
<span id="134">134</span>
<span id="135">135</span>
<span id="136">136</span>
<span id="137">137</span>
<span id="138">138</span>
<span id="139">139</span>
<span id="140">140</span>
<span id="141">141</span>
</pre><pre class="rust"><code><span class="kw">use</span> <span class="ident">std::collections::VecDeque</span>;
<span class="kw">use</span> <span class="ident">std::io</span>;
<span class="kw">use</span> <span class="ident">actix::prelude</span>::<span class="kw-2">*</span>;
<span class="kw">use</span> <span class="ident">actix_rt::net::TcpStream</span>;
<span class="kw">use</span> <span class="ident">actix_service::boxed</span>::{<span class="self">self</span>, <span class="ident">BoxService</span>};
<span class="kw">use</span> <span class="ident">actix_tls::connect</span>::{<span class="ident">ConnectError</span>, <span class="ident">ConnectInfo</span>, <span class="ident">Connection</span>, <span class="ident">ConnectorService</span>};
<span class="kw">use</span> <span class="ident">backoff::backoff::Backoff</span>;
<span class="kw">use</span> <span class="ident">backoff::ExponentialBackoff</span>;
<span class="kw">use</span> <span class="ident">log</span>::{<span class="ident">error</span>, <span class="ident">info</span>, <span class="ident">warn</span>};
<span class="kw">use</span> <span class="ident">redis_async::error::Error</span> <span class="kw">as</span> <span class="ident">RespError</span>;
<span class="kw">use</span> <span class="ident">redis_async::resp</span>::{<span class="ident">RespCodec</span>, <span class="ident">RespValue</span>};
<span class="kw">use</span> <span class="ident">tokio::io</span>::{<span class="ident">split</span>, <span class="ident">WriteHalf</span>};
<span class="kw">use</span> <span class="ident">tokio::sync::oneshot</span>;
<span class="kw">use</span> <span class="ident">tokio_util::codec::FramedRead</span>;
<span class="kw">use</span> <span class="ident"><span class="kw">crate</span>::Error</span>;
<span class="doccomment">/// Command for send data to Redis</span>
<span class="attribute">#[<span class="ident">derive</span>(<span class="ident">Debug</span>)]</span>
<span class="kw">pub</span> <span class="kw">struct</span> <span class="ident">Command</span>(<span class="kw">pub</span> <span class="ident">RespValue</span>);
<span class="kw">impl</span> <span class="ident">Message</span> <span class="kw">for</span> <span class="ident">Command</span> {
<span class="kw">type</span> <span class="prelude-ty">Result</span> <span class="op">=</span> <span class="prelude-ty">Result</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">Error</span><span class="op">&gt;</span>;
}
<span class="doccomment">/// Redis communication actor</span>
<span class="kw">pub</span> <span class="kw">struct</span> <span class="ident">RedisActor</span> {
<span class="ident">addr</span>: <span class="ident">String</span>,
<span class="ident">connector</span>: <span class="ident">BoxService</span><span class="op">&lt;</span><span class="ident">ConnectInfo</span><span class="op">&lt;</span><span class="ident">String</span><span class="op">&gt;</span>, <span class="ident">Connection</span><span class="op">&lt;</span><span class="ident">String</span>, <span class="ident">TcpStream</span><span class="op">&gt;</span>, <span class="ident">ConnectError</span><span class="op">&gt;</span>,
<span class="ident">backoff</span>: <span class="ident">ExponentialBackoff</span>,
<span class="ident">cell</span>: <span class="prelude-ty">Option</span><span class="op">&lt;</span><span class="ident">actix::io::FramedWrite</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">WriteHalf</span><span class="op">&lt;</span><span class="ident">TcpStream</span><span class="op">&gt;</span>, <span class="ident">RespCodec</span><span class="op">&gt;</span><span class="op">&gt;</span>,
<span class="ident">queue</span>: <span class="ident">VecDeque</span><span class="op">&lt;</span><span class="ident">oneshot::Sender</span><span class="op">&lt;</span><span class="prelude-ty">Result</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">Error</span><span class="op">&gt;</span><span class="op">&gt;</span><span class="op">&gt;</span>,
}
<span class="kw">impl</span> <span class="ident">RedisActor</span> {
<span class="doccomment">/// Start new `Supervisor` with `RedisActor`.</span>
<span class="kw">pub</span> <span class="kw">fn</span> <span class="ident">start</span><span class="op">&lt;</span><span class="ident">S</span>: <span class="ident">Into</span><span class="op">&lt;</span><span class="ident">String</span><span class="op">&gt;</span><span class="op">&gt;</span>(<span class="ident">addr</span>: <span class="ident">S</span>) -&gt; <span class="ident">Addr</span><span class="op">&lt;</span><span class="ident">RedisActor</span><span class="op">&gt;</span> {
<span class="kw">let</span> <span class="ident">addr</span> <span class="op">=</span> <span class="ident">addr</span>.<span class="ident">into</span>();
<span class="kw">let</span> <span class="ident">backoff</span> <span class="op">=</span> <span class="ident">ExponentialBackoff</span> {
<span class="ident">max_elapsed_time</span>: <span class="prelude-val">None</span>,
..<span class="ident">Default::default</span>()
};
<span class="ident">Supervisor::start</span>(<span class="op">|</span><span class="kw">_</span><span class="op">|</span> <span class="ident">RedisActor</span> {
<span class="ident">addr</span>,
<span class="ident">connector</span>: <span class="ident">boxed::service</span>(<span class="ident">ConnectorService::default</span>()),
<span class="ident">cell</span>: <span class="prelude-val">None</span>,
<span class="ident">backoff</span>,
<span class="ident">queue</span>: <span class="ident">VecDeque::new</span>(),
})
}
}
<span class="kw">impl</span> <span class="ident">Actor</span> <span class="kw">for</span> <span class="ident">RedisActor</span> {
<span class="kw">type</span> <span class="ident">Context</span> <span class="op">=</span> <span class="ident">Context</span><span class="op">&lt;</span><span class="self">Self</span><span class="op">&gt;</span>;
<span class="kw">fn</span> <span class="ident">started</span>(<span class="kw-2">&amp;mut</span> <span class="self">self</span>, <span class="ident">ctx</span>: <span class="kw-2">&amp;mut</span> <span class="ident">Context</span><span class="op">&lt;</span><span class="self">Self</span><span class="op">&gt;</span>) {
<span class="kw">let</span> <span class="ident">req</span> <span class="op">=</span> <span class="ident">ConnectInfo::new</span>(<span class="self">self</span>.<span class="ident">addr</span>.<span class="ident">to_owned</span>());
<span class="self">self</span>.<span class="ident">connector</span>
.<span class="ident">call</span>(<span class="ident">req</span>)
.<span class="ident">into_actor</span>(<span class="self">self</span>)
.<span class="ident">map</span>(<span class="op">|</span><span class="ident">res</span>, <span class="ident">act</span>, <span class="ident">ctx</span><span class="op">|</span> <span class="kw">match</span> <span class="ident">res</span> {
<span class="prelude-val">Ok</span>(<span class="ident">conn</span>) =&gt; {
<span class="kw">let</span> <span class="ident">stream</span> <span class="op">=</span> <span class="ident">conn</span>.<span class="ident">into_parts</span>().<span class="number">0</span>;
<span class="macro">info!</span>(<span class="string">&quot;Connected to redis server: {}&quot;</span>, <span class="ident">act</span>.<span class="ident">addr</span>);
<span class="kw">let</span> (<span class="ident">r</span>, <span class="ident">w</span>) <span class="op">=</span> <span class="ident">split</span>(<span class="ident">stream</span>);
<span class="comment">// configure write side of the connection</span>
<span class="kw">let</span> <span class="ident">framed</span> <span class="op">=</span> <span class="ident">actix::io::FramedWrite::new</span>(<span class="ident">w</span>, <span class="ident">RespCodec</span>, <span class="ident">ctx</span>);
<span class="ident">act</span>.<span class="ident">cell</span> <span class="op">=</span> <span class="prelude-val">Some</span>(<span class="ident">framed</span>);
<span class="comment">// read side of the connection</span>
<span class="ident">ctx</span>.<span class="ident">add_stream</span>(<span class="ident">FramedRead::new</span>(<span class="ident">r</span>, <span class="ident">RespCodec</span>));
<span class="ident">act</span>.<span class="ident">backoff</span>.<span class="ident">reset</span>();
}
<span class="prelude-val">Err</span>(<span class="ident">err</span>) =&gt; {
<span class="macro">error!</span>(<span class="string">&quot;Can not connect to redis server: {}&quot;</span>, <span class="ident">err</span>);
<span class="comment">// re-connect with backoff time.</span>
<span class="comment">// we stop current context, supervisor will restart it.</span>
<span class="kw">if</span> <span class="kw">let</span> <span class="prelude-val">Some</span>(<span class="ident">timeout</span>) <span class="op">=</span> <span class="ident">act</span>.<span class="ident">backoff</span>.<span class="ident">next_backoff</span>() {
<span class="ident">ctx</span>.<span class="ident">run_later</span>(<span class="ident">timeout</span>, <span class="op">|</span><span class="kw">_</span>, <span class="ident">ctx</span><span class="op">|</span> <span class="ident">ctx</span>.<span class="ident">stop</span>());
}
}
})
.<span class="ident">wait</span>(<span class="ident">ctx</span>);
}
}
<span class="kw">impl</span> <span class="ident">Supervised</span> <span class="kw">for</span> <span class="ident">RedisActor</span> {
<span class="kw">fn</span> <span class="ident">restarting</span>(<span class="kw-2">&amp;mut</span> <span class="self">self</span>, <span class="kw">_</span>: <span class="kw-2">&amp;mut</span> <span class="ident"><span class="self">Self</span>::Context</span>) {
<span class="self">self</span>.<span class="ident">cell</span>.<span class="ident">take</span>();
<span class="kw">for</span> <span class="ident">tx</span> <span class="kw">in</span> <span class="self">self</span>.<span class="ident">queue</span>.<span class="ident">drain</span>(..) {
<span class="kw">let</span> <span class="kw">_</span> <span class="op">=</span> <span class="ident">tx</span>.<span class="ident">send</span>(<span class="prelude-val">Err</span>(<span class="ident">Error::Disconnected</span>));
}
}
}
<span class="kw">impl</span> <span class="ident">actix::io::WriteHandler</span><span class="op">&lt;</span><span class="ident">io::Error</span><span class="op">&gt;</span> <span class="kw">for</span> <span class="ident">RedisActor</span> {
<span class="kw">fn</span> <span class="ident">error</span>(<span class="kw-2">&amp;mut</span> <span class="self">self</span>, <span class="ident">err</span>: <span class="ident">io::Error</span>, <span class="kw">_</span>: <span class="kw-2">&amp;mut</span> <span class="ident"><span class="self">Self</span>::Context</span>) -&gt; <span class="ident">Running</span> {
<span class="macro">warn!</span>(<span class="string">&quot;Redis connection dropped: {} error: {}&quot;</span>, <span class="self">self</span>.<span class="ident">addr</span>, <span class="ident">err</span>);
<span class="ident">Running::Stop</span>
}
}
<span class="kw">impl</span> <span class="ident">StreamHandler</span><span class="op">&lt;</span><span class="prelude-ty">Result</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">RespError</span><span class="op">&gt;</span><span class="op">&gt;</span> <span class="kw">for</span> <span class="ident">RedisActor</span> {
<span class="kw">fn</span> <span class="ident">handle</span>(<span class="kw-2">&amp;mut</span> <span class="self">self</span>, <span class="ident">msg</span>: <span class="prelude-ty">Result</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">RespError</span><span class="op">&gt;</span>, <span class="ident">ctx</span>: <span class="kw-2">&amp;mut</span> <span class="ident"><span class="self">Self</span>::Context</span>) {
<span class="kw">match</span> <span class="ident">msg</span> {
<span class="prelude-val">Err</span>(<span class="ident">e</span>) =&gt; {
<span class="kw">if</span> <span class="kw">let</span> <span class="prelude-val">Some</span>(<span class="ident">tx</span>) <span class="op">=</span> <span class="self">self</span>.<span class="ident">queue</span>.<span class="ident">pop_front</span>() {
<span class="kw">let</span> <span class="kw">_</span> <span class="op">=</span> <span class="ident">tx</span>.<span class="ident">send</span>(<span class="prelude-val">Err</span>(<span class="ident">e</span>.<span class="ident">into</span>()));
}
<span class="ident">ctx</span>.<span class="ident">stop</span>();
}
<span class="prelude-val">Ok</span>(<span class="ident">val</span>) =&gt; {
<span class="kw">if</span> <span class="kw">let</span> <span class="prelude-val">Some</span>(<span class="ident">tx</span>) <span class="op">=</span> <span class="self">self</span>.<span class="ident">queue</span>.<span class="ident">pop_front</span>() {
<span class="kw">let</span> <span class="kw">_</span> <span class="op">=</span> <span class="ident">tx</span>.<span class="ident">send</span>(<span class="prelude-val">Ok</span>(<span class="ident">val</span>));
}
}
}
}
}
<span class="kw">impl</span> <span class="ident">Handler</span><span class="op">&lt;</span><span class="ident">Command</span><span class="op">&gt;</span> <span class="kw">for</span> <span class="ident">RedisActor</span> {
<span class="kw">type</span> <span class="prelude-ty">Result</span> <span class="op">=</span> <span class="ident">ResponseFuture</span><span class="op">&lt;</span><span class="prelude-ty">Result</span><span class="op">&lt;</span><span class="ident">RespValue</span>, <span class="ident">Error</span><span class="op">&gt;</span><span class="op">&gt;</span>;
<span class="kw">fn</span> <span class="ident">handle</span>(<span class="kw-2">&amp;mut</span> <span class="self">self</span>, <span class="ident">msg</span>: <span class="ident">Command</span>, <span class="kw">_</span>: <span class="kw-2">&amp;mut</span> <span class="ident"><span class="self">Self</span>::Context</span>) -&gt; <span class="ident"><span class="self">Self</span>::Result</span> {
<span class="kw">let</span> (<span class="ident">tx</span>, <span class="ident">rx</span>) <span class="op">=</span> <span class="ident">oneshot::channel</span>();
<span class="kw">if</span> <span class="kw">let</span> <span class="prelude-val">Some</span>(<span class="kw-2">ref</span> <span class="kw-2">mut</span> <span class="ident">cell</span>) <span class="op">=</span> <span class="self">self</span>.<span class="ident">cell</span> {
<span class="self">self</span>.<span class="ident">queue</span>.<span class="ident">push_back</span>(<span class="ident">tx</span>);
<span class="ident">cell</span>.<span class="ident">write</span>(<span class="ident">msg</span>.<span class="number">0</span>);
} <span class="kw">else</span> {
<span class="kw">let</span> <span class="kw">_</span> <span class="op">=</span> <span class="ident">tx</span>.<span class="ident">send</span>(<span class="prelude-val">Err</span>(<span class="ident">Error::NotConnected</span>));
}
<span class="ident">Box::pin</span>(<span class="kw">async</span> <span class="kw">move</span> { <span class="ident">rx</span>.<span class="kw">await</span>.<span class="ident">map_err</span>(<span class="op">|</span><span class="kw">_</span><span class="op">|</span> <span class="ident">Error::Disconnected</span>)<span class="question-mark">?</span> })
}
}
</code></pre></div>
</section><section id="search" class="content hidden"></section></div></main><div id="rustdoc-vars" data-root-path="../../" data-current-crate="actix_redis" data-themes="ayu,dark,light" data-resource-suffix="" data-rustdoc-version="1.61.0-nightly (4b043faba 2022-02-24)" ></div>
</body></html>
Reference in New Issue View Git Blame Copy Permalink