- completed IBeanstalkClient implementation

- added some IBeanstalkClient extension methods for string payloads

Author: sdether

Beanstalk.Client/Beanstalk.Client.csproj

@@ -47,6 +47,8 @@
   <ItemGroup>
     <Compile Include="BeanstalkClient.cs" />
     <Compile Include="BeanstalkClientExceptions.cs" />
+    <Compile Include="BeanstalkDefaults.cs" />
+    <Compile Include="BeanstalkExtensions.cs" />
     <Compile Include="Protocol\ISocket.cs" />
     <Compile Include="Protocol\IWatchedTubeClient.cs" />
     <Compile Include="IWatchedTubeCollection.cs" />
@@ -63,6 +65,7 @@
     <Compile Include="PutResponse.cs" />
     <Compile Include="ReleaseStatus.cs" />
     <Compile Include="ServerStats.cs" />
+    <Compile Include="Protocol\StatsBase.cs" />
     <Compile Include="TubeStats.cs" />
     <Compile Include="Protocol\Request.cs" />
     <Compile Include="Protocol\RequestData.cs" />

Beanstalk.Client/BeanstalkClient.cs

@@ -29,14 +29,16 @@ namespace Droog.Beanstalk.Client {
 
     // TODO: should query the current and watched tubes at start-up, especially once there are connection pools
     public class BeanstalkClient : IBeanstalkClient, IWatchedTubeClient {
+
         public static readonly TimeSpan DefaultConnectTimeout = TimeSpan.FromSeconds(10);
 
         private readonly Func<ISocket> _socketFactory;
+        private readonly byte[] _buffer = new byte[16 * 1024];
+        private readonly TubeCollectionProxy _watchedTubes;
+        private readonly BeanstalkDefaults _defaults = new BeanstalkDefaults();
         private ISocket _socket;
         private bool _isDisposed;
-        private readonly byte[] _buffer = new byte[16 * 1024];
         private string _currentTube = "default";
-        private readonly TubeCollectionProxy _watchedTubes;
 
         public BeanstalkClient(IPAddress address, int port)
             : this(address, port, DefaultConnectTimeout) {
@@ -127,6 +129,10 @@ public void Connect() {
             VerifyConnection();
         }
 
+        public BeanstalkDefaults Defaults {
+            get { return _defaults; }
+        }
+
         public PutResponse Put(uint priority, TimeSpan delay, TimeSpan timeToRun, Stream request, long length) {
             var response = Exec(Request.Create(RequestCommand.Put)
                 .AppendArgument(priority)
@@ -183,39 +189,51 @@ public ReleaseStatus Release(uint jobId, uint priority, TimeSpan delay) {
         }
 
         public bool Bury(uint jobId, uint priority) {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.Bury)
+                .AppendArgument(jobId)
+                .AppendArgument(priority)
+                .ExpectStatuses(ResponseStatus.Buried | ResponseStatus.NotFound));
+            return response.Status == ResponseStatus.Buried;
         }
 
         public bool Touch(uint jobId) {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.Touch)
+                .AppendArgument(jobId)
+                .ExpectStatuses(ResponseStatus.Touched | ResponseStatus.NotFound));
+            return response.Status == ResponseStatus.Touched;
         }
 
-        public PeekResponse Peek(uint jobId) {
-            throw new NotImplementedException();
+        public Job Peek(uint jobId) {
+            return Peek(Request.Create(RequestCommand.Peek).AppendArgument(jobId));
         }
 
-        public PeekResponse PeekReady() {
-            throw new NotImplementedException();
+        public Job PeekReady() {
+            return Peek(Request.Create(RequestCommand.PeekReady));
         }
 
-        public PeekResponse PeekDelayed() {
-            throw new NotImplementedException();
+        public Job PeekDelayed() {
+            return Peek(Request.Create(RequestCommand.PeekDelayed));
         }
 
-        public PeekResponse PeekBuried() {
-            throw new NotImplementedException();
+        public Job PeekBuried() {
+            return Peek(Request.Create(RequestCommand.PeekBuried));
         }
 
         public uint Kick(uint bound) {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.Touch)
+                .AppendArgument(bound)
+                .ExpectStatuses(ResponseStatus.Kicked));
+            return uint.Parse(response.Arguments[0]);
         }
 
         public JobStats GetJobStats(uint jobId) {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.StatsJob).ExpectStatuses(ResponseStatus.Ok));
+            return new JobStats(MicroYaml.ParseDictionary(response));
         }
 
         public TubeStats GetTubeStats(string tube) {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.StatsTube).ExpectStatuses(ResponseStatus.Ok));
+            return new TubeStats(MicroYaml.ParseDictionary(response));
         }
 
         public ServerStats GetServerStats() {
@@ -224,7 +242,8 @@ public ServerStats GetServerStats() {
         }
 
         public IEnumerable<string> GetTubes() {
-            throw new NotImplementedException();
+            var response = Exec(Request.Create(RequestCommand.ListTubes).ExpectStatuses(ResponseStatus.Ok));
+            return MicroYaml.ParseList(response);
         }
 
         public void Close() {
@@ -253,6 +272,13 @@ IEnumerable<string> IWatchedTubeClient.ListWatchedTubes() {
             return MicroYaml.ParseList(response);
         }
 
+        private Job Peek(Request request) {
+            var response = Exec(request.ExpectStatuses(ResponseStatus.Found | ResponseStatus.NotFound));
+            return response.Status == ResponseStatus.NotFound 
+                       ? null 
+                       : new Job(uint.Parse(response.Arguments[0]), response.Data, long.Parse(response.Arguments[1]));
+        }
+
         private Response Exec(Request request) {
             VerifyConnection();
             _socket.SendRequest(request, _buffer);

Beanstalk.Client/BeanstalkDefaults.cs

@@ -0,0 +1,28 @@
+/*
+ * libBeanstalk.NET 
+ * Copyright (C) 2010 Arne F. Claassen
+ * geekblog [at] claassen [dot] net
+ * http://github.com/sdether/libBeanstalk.NET 
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+
+namespace Droog.Beanstalk.Client {
+    public class BeanstalkDefaults {
+        public uint Priority = ushort.MaxValue;
+        public TimeSpan Delay = TimeSpan.Zero;
+        public TimeSpan TimeToRun = TimeSpan.FromMinutes(10);
+    }
+}

Beanstalk.Client/BeanstalkExtensions.cs

@@ -0,0 +1,68 @@
+/*
+ * libBeanstalk.NET 
+ * Copyright (C) 2010 Arne F. Claassen
+ * geekblog [at] claassen [dot] net
+ * http://github.com/sdether/libBeanstalk.NET 
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+using System;
+using System.IO;
+using System.Text;
+
+namespace Droog.Beanstalk.Client {
+    public static class BeanstalkExtensions {
+
+        public static PutResponse Put(this IBeanstalkClient client, string data) {
+            using(var stream = new MemoryStream(Encoding.UTF8.GetBytes(data))) {
+                return client.Put(client.Defaults.Priority, client.Defaults.Delay, client.Defaults.TimeToRun, stream, stream.Length);
+            }
+        }
+
+        public static PutResponse Put(this IBeanstalkClient client, string data, uint priority) {
+            using(var stream = new MemoryStream(Encoding.UTF8.GetBytes(data))) {
+                return client.Put(priority, client.Defaults.Delay, client.Defaults.TimeToRun, stream, stream.Length);
+            }
+        }
+
+        public static PutResponse Put(this IBeanstalkClient client, string data, uint priority, TimeSpan delay) {
+            using(var stream = new MemoryStream(Encoding.UTF8.GetBytes(data))) {
+                return client.Put(priority, delay, client.Defaults.TimeToRun, stream, stream.Length);
+            }
+        }
+
+        public static PutResponse Put(this IBeanstalkClient client, string data, uint priority, TimeSpan delay, TimeSpan timeToRun) {
+            using(var stream = new MemoryStream(Encoding.UTF8.GetBytes(data))) {
+                return client.Put(priority, delay, timeToRun, stream, stream.Length);
+            }
+        }
+
+        public static Job<string> Reserve(this IBeanstalkClient client) {
+            var job = client.Reserve();
+            using(var reader = new StreamReader(job.Data)) {
+                return new Job<string>(job.JobId, reader.ReadToEnd());
+            }
+        }
+
+        public static Job<string> Reserve(this IBeanstalkClient client, TimeSpan timeout) {
+            var job = client.Reserve(timeout);
+            if(job == null) {
+                return null;
+            }
+            using(var reader = new StreamReader(job.Data)) {
+                return new Job<string>(job.JobId, reader.ReadToEnd());
+            }
+        }
+    }
+}

Beanstalk.Client/IBeanstalkClient.cs

@@ -35,6 +35,8 @@ public interface IBeanstalkClient : IDisposable
         // Consumer related
         string CurrentTube { get; set; }
 
+        BeanstalkDefaults Defaults { get; }
+
         PutResponse Put(uint priority, TimeSpan delay, TimeSpan timeToRun, Stream request, long length);
 
         // Producer related
@@ -48,10 +50,10 @@ public interface IBeanstalkClient : IDisposable
         bool Touch(uint jobId);
 
         // Misc
-        PeekResponse Peek(uint jobId);
-        PeekResponse PeekReady();
-        PeekResponse PeekDelayed();
-        PeekResponse PeekBuried();
+        Job Peek(uint jobId);
+        Job PeekReady();
+        Job PeekDelayed();
+        Job PeekBuried();
         uint Kick(uint bound);
         JobStats GetJobStats(uint jobId);
         TubeStats GetTubeStats(string tube);

Beanstalk.Client/Job.cs

@@ -18,6 +18,7 @@
  */
 
 using System.IO;
+using System.Text;
 
 namespace Droog.Beanstalk.Client {
     public class Job {
@@ -35,4 +36,19 @@ public Job(uint jobId, Stream data, long length) {
         public Stream Data { get { return _data; } }
         public long DataLength { get { return _length; } }
     }
+
+    public class Job<T> {
+        private readonly uint _jobId;
+        private readonly T _data;
+
+        public Job(uint jobId, T data) {
+            _jobId = jobId;
+            _data = data;
+        }
+
+        public uint JobId { get { return _jobId; } }
+        public T Data { get { return _data; } }
+    }
+
+
 }

Beanstalk.Client/JobStats.cs

@@ -17,6 +17,11 @@
  * limitations under the License.
  */
 
+using System.Collections.Generic;
+using Droog.Beanstalk.Client.Protocol;
+
 namespace Droog.Beanstalk.Client {
-    public class JobStats {}
+    public class JobStats : StatsBase {
+        public JobStats(IDictionary<string, string> dictionary) : base(dictionary) { }
+    }
 }

Beanstalk.Client/Protocol/StatsBase.cs

@@ -0,0 +1,35 @@
+/*
+ * libBeanstalk.NET 
+ * Copyright (C) 2010 Arne F. Claassen
+ * geekblog [at] claassen [dot] net
+ * http://github.com/sdether/libBeanstalk.NET 
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+using System.Collections.Generic;
+
+namespace Droog.Beanstalk.Client.Protocol {
+    public abstract class StatsBase {
+        protected readonly IDictionary<string, string> _dictionary;
+
+        protected StatsBase(IDictionary<string, string> dictionary) { _dictionary = dictionary; }
+
+        public string this[string key] {
+            get {
+                string value;
+                _dictionary.TryGetValue(key, out value);
+                return value;
+            }
+        }
+    }
+}

Beanstalk.Client/ServerStats.cs

@@ -18,21 +18,13 @@
  */
 
 using System.Collections.Generic;
+using Droog.Beanstalk.Client.Protocol;
 
 namespace Droog.Beanstalk.Client {
-    public class ServerStats {
-        private readonly IDictionary<string, string> _dictionary;
+    public class ServerStats : StatsBase {
 
-        public ServerStats(IDictionary<string, string> dictionary) {
-            _dictionary = dictionary;
-        }
-
-        public string this[string key] {
-            get {
-                string value;
-                _dictionary.TryGetValue(key, out value);
-                return value;
-            }
+        public ServerStats(IDictionary<string, string> dictionary)
+            : base(dictionary) {
         }
     }
 }

Beanstalk.Client/TubeStats.cs

@@ -17,6 +17,11 @@
  * limitations under the License.
  */
 
+using System.Collections.Generic;
+using Droog.Beanstalk.Client.Protocol;
+
 namespace Droog.Beanstalk.Client {
-    public class TubeStats {}
+    public class TubeStats : StatsBase {
+        public TubeStats(IDictionary<string, string> dictionary) : base(dictionary) { }
+    }
 }