1 /** 2 * PostgreSQL protocol implementation. 3 * !!! UNFINISHED !!! 4 * 5 * License: 6 * This Source Code Form is subject to the terms of 7 * the Mozilla Public License, v. 2.0. If a copy of 8 * the MPL was not distributed with this file, You 9 * can obtain one at http://mozilla.org/MPL/2.0/. 10 * 11 * Authors: 12 * Vladimir Panteleev <ae@cy.md> 13 */ 14 15 module ae.net.db.psql; 16 17 import std.array; 18 import std.exception; 19 import std.string; 20 21 import std.bitmanip : nativeToBigEndian, bigEndianToNative; 22 23 import ae.net.asockets; 24 import ae.utils.array; 25 import ae.utils.exception; 26 27 class PgSqlConnection 28 { 29 public: 30 this(IConnection conn, string user, string database) 31 { 32 this.conn = conn; 33 this.user = user; 34 this.database = database; 35 36 conn.handleConnect = &onConnect; 37 conn.handleReadData = &onReadData; 38 } 39 40 struct ErrorResponse 41 { 42 struct Field 43 { 44 char type; 45 char[] str; 46 47 string toString() { return "%s=%s".format(type, str); } 48 } 49 Field[] fields; 50 51 string toString() 52 { 53 return "%-(%s;%)".format(fields); 54 } 55 } 56 57 enum TransactionStatus : char 58 { 59 idle = 'I', 60 inTransaction = 'T', 61 failed = 'E', 62 } 63 64 struct FieldDescription 65 { 66 char[] name; 67 uint tableID; 68 uint type; 69 short size; 70 uint modifier; 71 ushort formatCode; 72 } 73 74 void delegate(ErrorResponse response) handleError; 75 void delegate() handleAuthenticated; 76 void delegate(char[] name, char[] value) handleParameterStatus; 77 void delegate(TransactionStatus transactionStatus) handleReadyForQuery; 78 79 string applicationName = "ae.net.db.psql"; 80 81 private: 82 IConnection conn; 83 84 string user; 85 string database; 86 87 enum ushort protocolVersionMajor = 3; 88 enum ushort protocolVersionMinor = 0; 89 90 enum PacketType : char 91 { 92 authenticationRequest = 'R', 93 backendKeyData = 'K', 94 errorResponse = 'E', 95 parameterStatus = 'S', 96 readyForQuery = 'Z', 97 rowDescription = 'T', 98 } 99 100 static T readInt(T)(ref Data data) 101 { 102 enforce!PgSqlException(data.length >= T.sizeof, "Not enough data in packet"); 103 return data.pop!(ubyte[T.sizeof]).bigEndianToNative!T(); 104 } 105 106 static char readChar(ref Data data) 107 { 108 return cast(char)readInt!ubyte(data); 109 } 110 111 static char[] readString(ref Data data) 112 { 113 char[] result; 114 data.asDataOf!char.enter((scope s) { 115 auto p = s.indexOf('\0'); 116 enforce!PgSqlException(p >= 0, "Unterminated string in packet"); 117 result = s[0..p].dup; 118 data = data[p+1..$]; 119 }); 120 return result; 121 } 122 123 void onConnect() 124 { 125 sendStartupMessage(); 126 } 127 128 Data packetBuf; 129 130 void onReadData(Data data) 131 { 132 packetBuf ~= data; 133 while (packetBuf.length >= 5) 134 { 135 auto length = { Data temp = packetBuf[1..5]; return readInt!uint(temp); }(); 136 if (packetBuf.length >= 1 + length) 137 { 138 auto packetData = packetBuf[0 .. 1 + length]; 139 packetBuf = packetBuf[1 + length .. $]; 140 if (!packetBuf.length) 141 packetBuf = Data.init; 142 143 auto packetType = cast(PacketType)readChar(packetData); 144 packetData = packetData[4..$]; // Skip length 145 processPacket(packetType, packetData); 146 } 147 } 148 } 149 150 void processPacket(PacketType type, Data data) 151 { 152 switch (type) 153 { 154 case PacketType.authenticationRequest: 155 { 156 auto result = readInt!uint(data); 157 enforce!PgSqlException(result == 0, "Authentication failed"); 158 if (handleAuthenticated) 159 handleAuthenticated(); 160 break; 161 } 162 case PacketType.backendKeyData: 163 { 164 // TODO? 165 break; 166 } 167 case PacketType.errorResponse: 168 { 169 ErrorResponse response; 170 while (data.length) 171 { 172 auto fieldType = readChar(data); 173 if (!fieldType) 174 break; 175 response.fields ~= ErrorResponse.Field(fieldType, readString(data)); 176 } 177 if (handleError) 178 handleError(response); 179 else 180 throw new PgSqlException(response.toString()); 181 break; 182 } 183 case PacketType.parameterStatus: 184 if (handleParameterStatus) 185 { 186 char[] name = readString(data); 187 char[] value = readString(data); 188 handleParameterStatus(name, value); 189 } 190 break; 191 case PacketType.readyForQuery: 192 if (handleReadyForQuery) 193 handleReadyForQuery(cast(TransactionStatus)readChar(data)); 194 break; 195 case PacketType.rowDescription: 196 { 197 auto fieldCount = readInt!ushort(data); 198 auto fields = new FieldDescription[fieldCount]; 199 foreach (n; 0..fieldCount) 200 { 201 } 202 break; 203 } 204 default: 205 throw new Exception("Unknown packet type '%s'".format(char(type))); 206 } 207 } 208 209 static void write(T)(ref Appender!(ubyte[]) buf, T value) 210 { 211 static if (is(T : long)) 212 { 213 buf.put(nativeToBigEndian(value)[]); 214 } 215 else 216 static if (is(T : const(char)[])) 217 { 218 buf.put(cast(const(ubyte)[])value); 219 buf.put(ubyte(0)); 220 } 221 else 222 static assert(false, "Can't write " ~ T.stringof); 223 } 224 225 void sendStartupMessage() 226 { 227 auto buf = appender!(ubyte[]); 228 229 write(buf, protocolVersionMajor); 230 write(buf, protocolVersionMinor); 231 232 write(buf, "user"); 233 write(buf, user); 234 235 write(buf, "database"); 236 write(buf, database); 237 238 write(buf, "application_name"); 239 write(buf, applicationName); 240 241 write(buf, "client_encoding"); 242 write(buf, "UTF8"); 243 244 write(buf, ""); 245 246 conn.send(Data(nativeToBigEndian(cast(uint)(buf.data.length + uint.sizeof))[])); 247 conn.send(Data(buf.data)); 248 } 249 250 void sendPacket(char type, const(ubyte)[] data) 251 { 252 conn.send(Data(type.asBytes)); 253 conn.send(Data(nativeToBigEndian(cast(uint)(data.length + uint.sizeof))[])); 254 conn.send(Data(data)); 255 } 256 257 void sendQuery(const(char)[] query) 258 { 259 auto buf = appender!(ubyte[]); 260 write(buf, query); 261 sendPacket('Q', buf.data); 262 } 263 } 264 265 mixin DeclareException!q{PgSqlException}; 266 267 version (HAVE_PSQL_SERVER) 268 unittest 269 { 270 import std.process : environment; 271 272 auto conn = new TcpConnection(); 273 auto pg = new PgSqlConnection(conn, environment["USER"], environment["USER"]); 274 conn.connect("localhost", 5432); 275 pg.handleReadyForQuery = (PgSqlConnection.TransactionStatus ts) { 276 pg.handleReadyForQuery = null; 277 pg.sendQuery("SELECT 2+2;"); 278 }; 279 socketManager.loop(); 280 }