Skip to content
This repository was archived by the owner on Jul 18, 2023. It is now read-only.

Commit 1c71d08

Browse files
gambrosenblumhardt
gambrose
authored and
nblumhardt
committed
Protocol writer (#36)
* Added tests for saving values of different types using line protocol point. * Convert LineProtocolPoint to use LineProtocolWriter * Clean up the client to accept one or many payloads. Added generic overload for struts only. * Removed unused LineProtocalSyntax class * Rename method after feedback in issue #35 * Removed ILineProtocolPayload to simplify api. * Reverted changes to LineProtocolPoint. Instead just added new advanced api without changing the old implementation. * Spelling fixes * Set position when timestamp written. * Simplified client api by not requiring creating the Writer using the client. Renamed start state for writer position. * Added tests for invalid writer states and exceptions produced. * Timestamp is last thing written in line so it does not make sense to allow fluent continuation from that. * Spelling * Added bit more context about what was trying to be written to exception message. * Ensure timestamps are UTC * Cannot single word * Fixed whitesapce in project files * Made write position private
1 parent 1a69c96 commit 1c71d08

5 files changed

Lines changed: 457 additions & 4 deletions

File tree

‎src/InfluxDB.LineProtocol/Client/LineProtocolClient.cs‎

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,15 +30,27 @@ protected LineProtocolClient(HttpMessageHandler handler, Uri serverBaseAddress,
3030
_password = password;
3131
}
3232

33-
public async Task<LineProtocolWriteResult> WriteAsync(LineProtocolPayload payload, CancellationToken cancellationToken = default(CancellationToken))
33+
public Task<LineProtocolWriteResult> WriteAsync(LineProtocolPayload payload, CancellationToken cancellationToken = default(CancellationToken))
34+
{
35+
var writer = new StringWriter();
36+
37+
payload.Format(writer);
38+
39+
return SendAsync(writer.ToString(), cancellationToken);
40+
}
41+
42+
public Task<LineProtocolWriteResult> SendAsync(LineProtocolWriter writer, CancellationToken cancellationToken = default(CancellationToken))
43+
{
44+
return SendAsync(writer.ToString(), cancellationToken);
45+
}
46+
47+
private async Task<LineProtocolWriteResult> SendAsync(string payload, CancellationToken cancellationToken = default(CancellationToken))
3448
{
3549
var endpoint = $"write?db={Uri.EscapeDataString(_database)}";
3650
if (!string.IsNullOrEmpty(_username))
3751
endpoint += $"&u={Uri.EscapeDataString(_username)}&p={Uri.EscapeDataString(_password)}";
3852

39-
var payloadText = new StringWriter();
40-
payload.Format(payloadText);
41-
var content = new StringContent(payloadText.ToString(), Encoding.UTF8);
53+
var content = new StringContent(payload, Encoding.UTF8);
4254
var response = await _httpClient.PostAsync(endpoint, content, cancellationToken).ConfigureAwait(false);
4355
if (response.IsSuccessStatusCode)
4456
{

‎src/InfluxDB.LineProtocol/LineProtocolWriter.cs‎

Lines changed: 239 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,239 @@
1+
using System;
2+
using System.Globalization;
3+
using System.IO;
4+
5+
namespace InfluxDB.LineProtocol
6+
{
7+
public class LineProtocolWriter
8+
{
9+
private static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
10+
11+
private readonly TextWriter textWriter;
12+
13+
private LinePosition position = LinePosition.NothingWritten;
14+
15+
public LineProtocolWriter()
16+
{
17+
this.textWriter = new StringWriter();
18+
}
19+
20+
public LineProtocolWriter Measurement(string name)
21+
{
22+
if (name == null)
23+
{
24+
throw new ArgumentNullException(nameof(name));
25+
}
26+
27+
if (string.IsNullOrWhiteSpace(name))
28+
{
29+
throw new ArgumentOutOfRangeException(nameof(name));
30+
}
31+
32+
switch (position)
33+
{
34+
case LinePosition.NothingWritten:
35+
break;
36+
case LinePosition.FieldWritten:
37+
case LinePosition.TimestampWritten:
38+
textWriter.Write("\n");
39+
break;
40+
default:
41+
throw InvalidPositionException($"Cannot write new measurement \"{name}\" as no field written for current line.");
42+
}
43+
44+
textWriter.Write(EscapeName(name));
45+
46+
position = LinePosition.MeasurementWritten;
47+
48+
return this;
49+
}
50+
51+
public LineProtocolWriter Tag(string name, string value)
52+
{
53+
switch (position)
54+
{
55+
case LinePosition.MeasurementWritten:
56+
case LinePosition.TagWritten:
57+
textWriter.Write(",");
58+
break;
59+
case LinePosition.NothingWritten:
60+
throw InvalidPositionException($"Cannot write tag \"{name}\" as no measurement name written.");
61+
default:
62+
throw InvalidPositionException($"Cannot write tag \"{name}\" as field(s) already written for current line.");
63+
}
64+
65+
textWriter.Write(EscapeName(name));
66+
textWriter.Write('=');
67+
textWriter.Write(EscapeName(value));
68+
69+
position = LinePosition.TagWritten;
70+
71+
return this;
72+
}
73+
74+
public LineProtocolWriter Field(string name, float value)
75+
{
76+
WriteFieldKey(name);
77+
textWriter.Write('=');
78+
textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
79+
80+
position = LinePosition.FieldWritten;
81+
82+
return this;
83+
}
84+
85+
public LineProtocolWriter Field(string name, double value)
86+
{
87+
WriteFieldKey(name);
88+
textWriter.Write('=');
89+
textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
90+
91+
position = LinePosition.FieldWritten;
92+
93+
return this;
94+
}
95+
96+
public LineProtocolWriter Field(string name, decimal value)
97+
{
98+
WriteFieldKey(name);
99+
textWriter.Write('=');
100+
textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
101+
102+
position = LinePosition.FieldWritten;
103+
104+
return this;
105+
}
106+
107+
public LineProtocolWriter Field(string name, long value)
108+
{
109+
WriteFieldKey(name);
110+
textWriter.Write('=');
111+
textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
112+
textWriter.Write('i');
113+
114+
position = LinePosition.FieldWritten;
115+
116+
return this;
117+
}
118+
119+
public LineProtocolWriter Field(string name, string value)
120+
{
121+
WriteFieldKey(name);
122+
textWriter.Write('=');
123+
textWriter.Write('"');
124+
textWriter.Write(value.Replace("\"", "\\\""));
125+
textWriter.Write('"');
126+
127+
position = LinePosition.FieldWritten;
128+
129+
return this;
130+
}
131+
132+
public LineProtocolWriter Field(string name, bool value)
133+
{
134+
WriteFieldKey(name);
135+
textWriter.Write('=');
136+
textWriter.Write(value ? 't' : 'f');
137+
138+
position = LinePosition.FieldWritten;
139+
140+
return this;
141+
}
142+
143+
public void Timestamp(long value)
144+
{
145+
switch (position)
146+
{
147+
case LinePosition.FieldWritten:
148+
textWriter.Write(" ");
149+
break;
150+
case LinePosition.NothingWritten:
151+
throw InvalidPositionException("Cannot write timestamp as no measurement name written.");
152+
default:
153+
throw InvalidPositionException("Cannot write timestamp as no field written for current measurement.");
154+
}
155+
156+
textWriter.Write(value.ToString(CultureInfo.InvariantCulture));
157+
158+
position = LinePosition.TimestampWritten;
159+
}
160+
161+
public void Timestamp(TimeSpan value)
162+
{
163+
Timestamp(value.Ticks * 100L);
164+
}
165+
166+
public void Timestamp(DateTimeOffset value)
167+
{
168+
Timestamp(value.UtcDateTime);
169+
}
170+
171+
public void Timestamp(DateTime value)
172+
{
173+
if (value != null && value.Kind != DateTimeKind.Utc)
174+
{
175+
throw new ArgumentException("Timestamps must be specified as UTC", nameof(value));
176+
}
177+
178+
Timestamp(value - UnixEpoch);
179+
}
180+
181+
public override string ToString()
182+
{
183+
return textWriter.ToString();
184+
}
185+
186+
private void WriteFieldKey(string name)
187+
{
188+
switch (position)
189+
{
190+
case LinePosition.MeasurementWritten:
191+
case LinePosition.TagWritten:
192+
textWriter.Write(" ");
193+
break;
194+
case LinePosition.FieldWritten:
195+
textWriter.Write(",");
196+
break;
197+
default:
198+
throw InvalidPositionException($"Cannot write field \"{name}\" as no measurement name written.");
199+
}
200+
201+
textWriter.Write(EscapeName(name));
202+
}
203+
204+
public static string EscapeName(string nameOrKey)
205+
{
206+
if (nameOrKey == null)
207+
{
208+
throw new ArgumentNullException(nameof(nameOrKey));
209+
}
210+
211+
return nameOrKey
212+
.Replace("=", "\\=")
213+
.Replace(" ", "\\ ")
214+
.Replace(",", "\\,");
215+
}
216+
217+
private InvalidOperationException InvalidPositionException(string message)
218+
{
219+
// We don't need an custom exceptions as there should be no need for a dev to catch this condition. They are not using the api right so how can the write code to recover.
220+
// We can make the current writer position available for better diagnostics then logged.
221+
return new InvalidOperationException(message)
222+
{
223+
Data =
224+
{
225+
{ "Position", position.ToString() }
226+
}
227+
};
228+
}
229+
230+
enum LinePosition
231+
{
232+
NothingWritten,
233+
MeasurementWritten,
234+
TagWritten,
235+
FieldWritten,
236+
TimestampWritten
237+
}
238+
}
239+
}

‎test/InfluxDB.LineProtocol.Tests/InfluxDB.LineProtocol.Tests.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616

1717
<ItemGroup>
1818
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="15.0.0" />
19+
<PackageReference Include="System.ValueTuple" Version="4.3.0" />
1920
<PackageReference Include="RichardSzalay.MockHttp" Version="1.5.0" />
2021
<PackageReference Include="xunit.runner.visualstudio" Version="2.2.0" />
2122
<PackageReference Include="xunit" Version="2.2.0" />

‎test/InfluxDB.LineProtocol.Tests/LineProtocolWriterInvalidStateTests.cs‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
using System;
2+
using Xunit;
3+
4+
namespace InfluxDB.LineProtocol.Tests
5+
{
6+
public class LineProtocolWriterInvalidStateTests
7+
{
8+
[Fact]
9+
public void Cannot_write_new_measurment_when_field_not_written()
10+
{
11+
var writer = new LineProtocolWriter();
12+
13+
writer.Measurement("my_measurement");
14+
15+
var ex = Assert.Throws<InvalidOperationException>(() => writer.Measurement("my_measurement"));
16+
Assert.Equal("Cannot write new measurement \"my_measurement\" as no field written for current line.", ex.Message);
17+
Assert.Equal("MeasurementWritten", ex.Data["Position"]);
18+
19+
writer.Tag("foo", "bar");
20+
21+
ex = Assert.Throws<InvalidOperationException>(() => writer.Measurement("my_measurement"));
22+
Assert.Equal("TagWritten", ex.Data["Position"]);
23+
}
24+
25+
[Fact]
26+
public void Cannot_write_tag_when_no_measurement_written()
27+
{
28+
var writer = new LineProtocolWriter();
29+
30+
var ex = Assert.Throws<InvalidOperationException>(() => writer.Tag("foo", "bar"));
31+
Assert.Equal("Cannot write tag \"foo\" as no measurement name written.", ex.Message);
32+
Assert.Equal("NothingWritten", ex.Data["Position"]);
33+
}
34+
35+
[Fact]
36+
public void Cannot_write_tag_after_field_written()
37+
{
38+
var writer = new LineProtocolWriter();
39+
40+
writer.Measurement("my_measurement").Field("value", 1);
41+
42+
var ex = Assert.Throws<InvalidOperationException>(() => writer.Tag("foo", "bar"));
43+
Assert.Equal("Cannot write tag \"foo\" as field(s) already written for current line.", ex.Message);
44+
Assert.Equal("FieldWritten", ex.Data["Position"]);
45+
46+
writer.Timestamp(123456);
47+
48+
ex = Assert.Throws<InvalidOperationException>(() => writer.Tag("foo", "bar"));
49+
Assert.Equal("TimestampWritten", ex.Data["Position"]);
50+
}
51+
52+
[Fact]
53+
public void Cannot_write_field_when_no_measurement_written()
54+
{
55+
var writer = new LineProtocolWriter();
56+
57+
var ex = Assert.Throws<InvalidOperationException>(() => writer.Field("value", 1.2));
58+
Assert.Equal("Cannot write field \"value\" as no measurement name written.", ex.Message);
59+
Assert.Equal("NothingWritten", ex.Data["Position"]);
60+
61+
Assert.Throws<InvalidOperationException>(() => writer.Field("value", 3));
62+
Assert.Throws<InvalidOperationException>(() => writer.Field("value", true));
63+
Assert.Throws<InvalidOperationException>(() => writer.Field("value", "roar like a lion"));
64+
Assert.Throws<InvalidOperationException>(() => writer.Field("value", 2.6f));
65+
Assert.Throws<InvalidOperationException>(() => writer.Field("value", 1000000m));
66+
}
67+
68+
[Fact]
69+
public void Cannot_write_timestamp_when_no_field_written()
70+
{
71+
var writer = new LineProtocolWriter();
72+
73+
var ex = Assert.Throws<InvalidOperationException>(() => writer.Timestamp(123456));
74+
Assert.Equal("Cannot write timestamp as no measurement name written.", ex.Message);
75+
Assert.Equal("NothingWritten", ex.Data["Position"]);
76+
77+
writer.Measurement("my_measurement");
78+
79+
ex = Assert.Throws<InvalidOperationException>(() => writer.Timestamp(123456));
80+
Assert.Equal("Cannot write timestamp as no field written for current measurement.", ex.Message);
81+
Assert.Equal("MeasurementWritten", ex.Data["Position"]);
82+
83+
writer.Tag("foo", "bar");
84+
85+
ex = Assert.Throws<InvalidOperationException>(() => writer.Timestamp(123456));
86+
Assert.Equal("TagWritten", ex.Data["Position"]);
87+
88+
Assert.Throws<InvalidOperationException>(() => writer.Timestamp(TimeSpan.FromDays(3045)));
89+
Assert.Throws<InvalidOperationException>(() => writer.Timestamp(DateTime.UtcNow));
90+
Assert.Throws<InvalidOperationException>(() => writer.Timestamp(DateTimeOffset.Now));
91+
}
92+
}
93+
}

0 commit comments

Comments
 (0)