vibenews.nntp.common source code

1 /**
2 	(module summary)
3 
4 	Copyright: © 2012-2014 RejectedSoftware e.K.
5 	License: Subject to the terms of the General Public License version 3, as written in the included LICENSE.txt file.
6 	Authors: Sönke Ludwig
7 */
8 module vibenews.nntp.common;
9 
10 import vibe.core.log;
11 import vibe.stream.operations;
12 
13 import std.algorithm;
14 import std.exception;
15 
16 
17 class NNTPBodyReader : InputStream {
18 	private {
19 		InputStreamProxy m_stream;
20 		bool m_eof = false;
21 		ubyte[] m_currLine;
22 	}
23 
24 	this(InputStreamProxy stream)
25 	{
26 		m_stream = stream;
27 		readNextLine();
28 	}
29 
30 	@property bool empty() { return m_eof; }
31 
32 	@property ulong leastSize() { return m_currLine.length; }
33 
34 	@property bool dataAvailableForRead(){
35 		return m_currLine.length > 0;
36 	}
37 
38 	@property ubyte[] peek()
39 	{
40 		return m_currLine;
41 	}
42 
43 	static if (!is(IOMode)) override void read(scope ubyte[] bytes) { readImpl(bytes); }
44 	else override size_t read(scope ubyte[] bytes, IOMode) { readImpl(bytes); return bytes.length; }
45 	alias read = InputStream.read;
46 
47 	private void readImpl()(scope ubyte[] dst)
48 	{
49 		while (dst.length > 0) {
50 			enforce(!m_eof);
51 			auto amt = min(dst.length, m_currLine.length);
52 			dst[0 .. amt] = m_currLine[0 .. amt];
53 			
54 			m_currLine = m_currLine[amt .. $];
55 			dst = dst[amt .. $];
56 			
57 			if( m_currLine.length == 0 )
58 				readNextLine();
59 		}
60 	}
61 
62 	private void readNextLine()()
63 	{
64 		enforce(!m_eof);
65 		m_currLine = m_stream.readLine() ~ cast(const(ubyte)[])"\r\n";
66 		m_eof = m_currLine == ".\r\n";
67 		if( m_currLine.startsWith("..") ) m_currLine = m_currLine[1 .. $];
68 	}
69 }
70 
71 deprecated alias NntpBodyReader = NNTPBodyReader;
72 
73 
74 class NNTPBodyWriter : OutputStream {
75 	private {
76 		OutputStreamProxy m_stream;
77 		bool m_finalized = false;
78 		int m_lineState = 0;
79 		static immutable ubyte[] m_lineStateString = cast(immutable(ubyte)[])"\r\n.";
80 		bool m_empty = true;
81 	}
82 
83 	this(OutputStreamProxy stream)
84 	{
85 		m_stream = stream;
86 	}
87 
88 	static if (!is(IOMode)) override void write(in ubyte[] bytes) { writeImpl(bytes); }
89 	else override size_t write(in ubyte[] bytes, IOMode) { writeImpl(bytes); return bytes.length; }
90 	alias write = OutputStream.write;
91 
92 	private void writeImpl()(in ubyte[] bytes_)
93 	{
94 		const(ubyte)[] bytes = bytes_;
95 		assert(!m_finalized);
96 
97 		if( bytes.length ){
98 			if( m_empty && bytes[0] == '.' ){
99 				m_stream.write("..");
100 				logDebugV("WS <..>", cast(const(char)[])bytes[0 .. $-m_lineState]);
101 				bytes = bytes[1 .. $];
102 			}
103 			m_empty = false;
104 		}
105 
106 		// test any already started prefix
107 		if( m_lineState > 0 ){
108 			foreach( i; m_lineState .. min(m_lineStateString.length, bytes.length+m_lineState) ){
109 				if( bytes[i-m_lineState] != m_lineStateString[i] ){
110 					m_stream.write(m_lineStateString[0 .. i]);
111 					bytes = bytes[i-m_lineState .. $];
112 					logDebugV("WPM <%s>", cast(const(char)[])m_lineStateString[0 .. i]);
113 					m_lineState = 0;
114 					break;
115 				}
116 			}
117 			if( m_lineState > 0 ){
118 				if( m_lineStateString.length > bytes.length+m_lineState ){
119 					m_lineState += bytes.length;
120 					bytes = null;
121 				} else {
122 					m_stream.write("\r\n..");
123 					logDebugV("WEM <\\r\\n..>");
124 					bytes = bytes[m_lineStateString.length-m_lineState .. $];
125 					m_lineState = 0;
126 				}
127 			}
128 		}
129 
130 		while( bytes.length ){
131 			auto idx = bytes.countUntil(m_lineStateString);
132 			if( idx >= 0 ){
133 				m_stream.write(bytes[0 .. idx]);
134 				m_stream.write("\r\n..");
135 				logDebugV("WMM <%s\\r\\n..>", cast(const(char)[])bytes[0 .. idx]);
136 				bytes = bytes[idx+m_lineStateString.length .. $];
137 			} else {
138 				foreach( i; 1 .. min(m_lineStateString.length, bytes.length) )
139 					if( bytes[$-i .. $] == m_lineStateString[0 .. i] ){
140 						m_lineState = cast(int)i;
141 						break;
142 					}
143 				m_stream.write(bytes[0 .. $-m_lineState]);
144 				logDebugV("WP <%s>", cast(const(char)[])bytes[0 .. $-m_lineState]);
145 				bytes = null;
146 			}
147 		}
148 	}
149 
150 	void flush()
151 	{
152 		m_stream.flush();
153 	}
154 
155 	void finalize()
156 	{
157 		if( m_lineState > 0 ) m_stream.write(m_lineStateString[0 .. m_lineState]);
158 		enforce(!m_finalized);
159 		m_finalized = true;
160 		if( m_empty ) m_stream.write(".\r\n");
161 		else m_stream.write("\r\n.\r\n");
162 		m_stream.flush();
163 		logDebugV("WF <\\r\\n.\\r\\n>");
164 	}
165 
166 	static if (!is(IOMode)) {
167 		override void write(InputStream stream, ulong nbytes = 0)
168 		{
169 			writeDefault(stream, nbytes);
170 		}
171 	}
172 }
173 
174 deprecated alias NntpBodyWriter = NNTPBodyReader;