This processes _queries from a client and dispatches the _queries to the database. More...

Inheritance diagram for Deveel.Data.Protocol.Processor:

Public Member Functions
abstract void	Close ()
	Closes the connection with the client. More...

void	Dispose ()

Protected Member Functions
	Processor (DbController controller, string hostString)

byte[]	ProcessCommand (byte[] command)
	Processes a single Query from the client. More...

void	Dispose (bool disposing)
	Disposes of this processor. More...

abstract void	SendEvent (byte[] eventMsg)
	Sends an event to the client. More...

Properties
ClientConnectionState	ClientState `[get]`
	Returns the state of the connection. More...

abstract bool	IsClosed `[get]`
	Returns true if the connection to the client is closed. More...

IDatabase	Database `[get, private set]`

Private Member Functions
void	OnDatabaseEvent (int eventType, String eventMessage)

bool	ChangeDatabaseInterface (IDbConfig config, string databaseName)

byte[]	ProcessQuery (byte[] command)
	Processes a query on the byte[] array and returns the result. More...

byte[]	ChangeDatabase (int dispatchId, byte[] command)

byte[]	QueryCommand (int dispatchId, byte[] command)
	Executes a query and returns the header for the result in the response. More...

byte[]	PushStreamableObjectPart (int dispatchId, byte[] command)
	Pushes a part of a streamable object onto the server. More...

byte[]	ResultSection (int dispatchId, byte[] command)
	Responds with a part of the result set of a query made via the ProtocolConstants.Query Query. More...

byte[]	StreamableObjectSection (int dispatchId, byte[] command)
	Returns a section of a streamable object. More...

byte[]	DisposeStreamableObject (int dispatchId, byte[] command)
	Disposes of a streamable object. More...

byte[]	DisposeResult (int dispatchId, byte[] command)
	Disposes of a result set we queries via the ProtocolConstants.Query Query. More...

	~Processor ()

Static Private Member Functions
static byte[]	Single (int val)
	Returns a single 4 byte array with the given int encoded into it. More...

static byte[]	Exception (int dispatchId, DataException e)
	Creates a response that represents a data exception failure. More...

static byte[]	SimpleSuccess (int dispatchId)
	Creates a response that indicates a simple success of an operation with the given dispatch id. More...

Private Attributes
const int	ServerVersion = 1
	The version of the server protocol. More...

ClientConnectionState	state
	The current state we are in. More...

int	authenticationTries
	Number of authentications tried. More...

readonly string	hostString

readonly DbController	controller

readonly Dictionary< string, DatabaseInterface >	dbInterfaces

DatabaseInterface	dbInterface

readonly DatabaseEventCallback	dbCallback
	The database call back method that sends database events back to the client. More...

Detailed Description

This processes _queries from a client and dispatches the _queries to the database.

This is a state based class. There is a single processor for each client connected. This class is designed to be flexible enough to handle packet based protocols as well as stream based protocols.

Definition at line 41 of file Processor.cs.

Constructor & Destructor Documentation

Deveel.Data.Protocol.Processor.Processor	(	DbController	controller,
		string	hostString
	)

inlineprotected

Definition at line 63 of file Processor.cs.

                                                                         {
             this.hostString = hostString;
             dbInterfaces = new Dictionary<string, DatabaseInterface>();
             this.controller = controller;
             state = 0;
             authenticationTries = 0;
             dbCallback = OnDatabaseEvent;
         }

Deveel.Data.Protocol.Processor.~Processor ( )

inlineprivate

Definition at line 554 of file Processor.cs.

                      {
             Dispose(false);
         }

Member Function Documentation

byte [] Deveel.Data.Protocol.Processor.ChangeDatabase	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Definition at line 319 of file Processor.cs.

                                                                        {
             // Read the query from the Query.
             MemoryStream input = new MemoryStream(command, 8, command.Length - 8);
             BinaryReader reader = new BinaryReader(input, Encoding.Unicode);
 
             string databaseName = reader.ReadString();
 
             try {
                 dbInterface.ChangeDatabase(databaseName);
                 state = ClientConnectionState.NotAuthenticated;
                 return SimpleSuccess(dispatchId);
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
         }

bool Deveel.Data.Protocol.Processor.ChangeDatabaseInterface	(	IDbConfig	config,
		string	databaseName
	)

inlineprivate

Definition at line 92 of file Processor.cs.

                                                                                     {
             if (!controller.DatabaseExists(config, databaseName))
                 return false;
 
             DatabaseInterface dbi;
             if (!dbInterfaces.TryGetValue(databaseName, out dbi)) {
                 dbi = new DatabaseInterface(controller, databaseName, hostString);
                 dbInterfaces[databaseName] = dbi;
             }
 
             Database = controller.GetDatabase(databaseName);
             dbInterface = dbi;
             return true;
         }

abstract void Deveel.Data.Protocol.Processor.Close ( )

pure virtual

Closes the connection with the client.

void Deveel.Data.Protocol.Processor.Dispose ( bool disposing )

inlineprotected

Disposes of this processor.

Definition at line 301 of file Processor.cs.

                                                {
             if (disposing) {
                 try {
                     if (dbInterfaces.Count > 0) {
                         foreach (DatabaseInterface databaseInterface in dbInterfaces.Values) {
                             databaseInterface.Dispose();
                         }
                     }
                     dbInterfaces.Clear();
                 } catch (Exception e) {
                     controller.Logger.Error(this, e);
                 }
             }
         }

void Deveel.Data.Protocol.Processor.Dispose ( )

inline

Definition at line 558 of file Processor.cs.

                               {
             Dispose(true);
             GC.SuppressFinalize(this);
         }

byte [] Deveel.Data.Protocol.Processor.DisposeResult	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Disposes of a result set we queries via the ProtocolConstants.Query Query.

Parameters

dispatchId
command

Returns

Definition at line 507 of file Processor.cs.

                                                                      {
             // Get the result id.
             int resultId = ByteBuffer.ReadInt4(command, 8);
 
             try {
                 // Dispose the table.
                 dbInterface.DisposeResult(resultId);
                 // Return operation success.
                 return SimpleSuccess(dispatchId);
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
         }

byte [] Deveel.Data.Protocol.Processor.DisposeStreamableObject	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Disposes of a streamable object.

Parameters

dispatchId
command

Returns

Definition at line 486 of file Processor.cs.

                                                                                {
             int resultId = ByteBuffer.ReadInt4(command, 8);
             long streamableObjectId = ByteBuffer.ReadInt8(command, 12);
 
             try {
                 // Pass this through to the underlying database interface.
                 dbInterface.DisposeStreamableObject(resultId, streamableObjectId);
 
                 // Return operation success.
                 return SimpleSuccess(dispatchId);
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
         }

static byte [] Deveel.Data.Protocol.Processor.Exception	(	int	dispatchId,
		DataException	e
	)

inlinestaticprivate

Creates a response that represents a data exception failure.

Parameters

dispatchId
e

Returns

Definition at line 214 of file Processor.cs.

                                                                          {
             int code = /* TODO: e.ErrorCode */ -1;
             string msg = e.Message;
             if (String.IsNullOrEmpty(msg))
                 msg = "NULL exception message";
 
             string server_msg = "";
             string stack_trace = "";
 
             if (e is DbDataException) {
                 DbDataException me = (DbDataException)e;
                 server_msg = me.ServerErrorMessage;
                 stack_trace = me.ServerErrorStackTrace;
             } else {
                 stack_trace = e.StackTrace;
             }
 
             MemoryStream output = new MemoryStream();
             BinaryWriter writer = new BinaryWriter(output, Encoding.Unicode);
             writer.Write(dispatchId);
             writer.Write(ProtocolConstants.Exception);
             writer.Write(code);
             writer.Write(msg);
             writer.Write(stack_trace);
 
             return output.ToArray();
         }

void Deveel.Data.Protocol.Processor.OnDatabaseEvent	(	int	eventType,
		String	eventMessage
	)

inlineprivate

Definition at line 72 of file Processor.cs.

                                                                          {
             try {
                 // Format the call back and send the event.
                 MemoryStream bout = new MemoryStream();
                 BinaryWriter dout = new BinaryWriter(bout, Encoding.Unicode);
                 dout.Write(eventType);
                 dout.Write(eventMessage);
                 SendEvent(bout.ToArray());
             } catch (IOException e) {
                 controller.Logger.Error(this, "IO Error: " + e.Message);
                 controller.Logger.Error(this, e);
             }
         }

byte [] Deveel.Data.Protocol.Processor.ProcessCommand ( byte[] command )

inlineprotected

Processes a single Query from the client.

Parameters

command

Returns: Returns a byte array and the response is written out as a byte array, or null if the connection has been closed.

Definition at line 115 of file Processor.cs.

                                                         {
             if (state == ClientConnectionState.Closed) {
                 // State 0 means we looking for the header...
                 BinaryReader reader = new BinaryReader(new MemoryStream(command), Encoding.ASCII);
                 /*
                 int magic = ByteBuffer.ReadInt4(Query, 0);
                 // The driver version number
                 int maj_ver = ByteBuffer.ReadInt4(Query, 4);
                 int min_ver = ByteBuffer.ReadInt4(Query, 8);
                 */
                 reader.ReadInt32();     // magic
                 reader.ReadInt32();     // server major version
                 reader.ReadInt32();     // server minor version
 
                 string databaseName = reader.ReadString();
 
                 if (!ChangeDatabaseInterface(controller.Config, databaseName))
                     return Single(ProtocolConstants.DatabaseNotFound);
 
                 Version version = Assembly.GetExecutingAssembly().GetName().Version;
 
                 byte[] ackCommand = new byte[4 + 1 + 4 + 4 + 1];
                 // Send back an acknowledgement and the version number of the server
                 ByteBuffer.WriteInteger(ProtocolConstants.Acknowledgement, ackCommand, 0);
                 ackCommand[4] = 1;
                 ByteBuffer.WriteInteger(version.Major, ackCommand, 5);
                 ByteBuffer.WriteInteger(version.Minor, ackCommand, 9);
                 ackCommand[13] = 0;
 
                 // Set to the next state.
                 state = ClientConnectionState.NotAuthenticated;
 
                 // Return the acknowledgement
                 return ackCommand;
             }
 
             if (state == ClientConnectionState.NotAuthenticated) {
                 // State 4 means we looking for username and password...
                 MemoryStream input = new MemoryStream(command);
                 BinaryReader reader = new BinaryReader(input, Encoding.ASCII);
                 string defaultSchema = reader.ReadString();
                 string username = reader.ReadString();
                 string password = reader.ReadString();
 
                 try {
                     if (!dbInterface.Login(defaultSchema, username, password, dbCallback)) {
                         // Close after 12 tries.
                         if (authenticationTries >= 12) {
                             Close();
                         } else {
                             ++authenticationTries;
                             return Single(ProtocolConstants.UserAuthenticationFailed);
                         }
                     } else {
                         state = ClientConnectionState.Processing;
                         return Single(ProtocolConstants.UserAuthenticationPassed);
                     }
                 } catch (DataException) {
 
                 }
 
                 return null;
             }
 
             if (state == ClientConnectionState.Processing)
                 // Process the query
                 return ProcessQuery(command);
 
             throw new Exception("Illegal state: " + state);
         }

byte [] Deveel.Data.Protocol.Processor.ProcessQuery ( byte[] command )

inlineprivate

Processes a query on the byte[] array and returns the result.

Parameters

command

Returns

Definition at line 260 of file Processor.cs.

                                                     {
             byte[] result;
 
             // The first int is the Query.
             int ins = ByteBuffer.ReadInt4(command, 0);
 
             // Otherwise must be a dispatch type request.
             // The second is the dispatch id.
             int dispatchId = ByteBuffer.ReadInt4(command, 4);
 
             if (dispatchId == -1)
                 throw new Exception("Special case dispatch id of -1 in query");
 
             if (ins == ProtocolConstants.ChangeDatabase) {
                 result = ChangeDatabase(dispatchId, command);
             } else if (ins == ProtocolConstants.ResultSection) {
                 result = ResultSection(dispatchId, command);
             } else if (ins == ProtocolConstants.Query) {
                 result = QueryCommand(dispatchId, command);
             } else if (ins == ProtocolConstants.PushStreamableObjectPart) {
                 result = PushStreamableObjectPart(dispatchId, command);
             } else if (ins == ProtocolConstants.DisposeResult) {
                 result = DisposeResult(dispatchId, command);
             } else if (ins == ProtocolConstants.StreamableObjectSection) {
                 result = StreamableObjectSection(dispatchId, command);
             } else if (ins == ProtocolConstants.DisposeStreamableObject) {
                 result = DisposeStreamableObject(dispatchId, command);
             } else if (ins == ProtocolConstants.Close) {
                 Close();
                 result = null;
             } else {
                 throw new Exception("Query (" + ins + ") not understood.");
             }
 
             return result;
 
         }

byte [] Deveel.Data.Protocol.Processor.PushStreamableObjectPart	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Pushes a part of a streamable object onto the server.

Parameters

dispatchId	The number we need to respond with.
command

Returns

Definition at line 389 of file Processor.cs.

                                                                                 {
             ReferenceType type = (ReferenceType) command[8];
             long objectId = ByteBuffer.ReadInt8(command, 9);
             long objectLength = ByteBuffer.ReadInt8(command, 17);
             int length = ByteBuffer.ReadInt4(command, 25);
             byte[] obBuf = new byte[length];
             Array.Copy(command, 29, obBuf, 0, length);
             long offset = ByteBuffer.ReadInt8(command, 29 + length);
 
             try {
                 // Pass this through to the underlying database interface.
                 dbInterface.PushStreamableObjectPart(type, objectId, objectLength, obBuf, offset, length);
 
                 // Return operation success.
                 return SimpleSuccess(dispatchId);
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
         }

byte [] Deveel.Data.Protocol.Processor.QueryCommand	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Executes a query and returns the header for the result in the response.

Parameters

dispatchId	The number we need to respond with.
command

This keeps track of all result sets because sections of the result are later queries via the ProtocolConstants.ResultSection Query.

Returns

Definition at line 346 of file Processor.cs.

                                                                     {
             // Read the query from the Query.
             MemoryStream input = new MemoryStream(command, 8, command.Length - 8);
             BinaryReader reader = new BinaryReader(input, Encoding.Unicode);
             SqlQuery query = SqlQuery.ReadFrom(reader);
 
             try {
                 // Do the query
                 IQueryResponse response = dbInterface.ExecuteQuery(query)[0];
 
                 // Prepare the stream to output the response to,
                 MemoryStream output = new MemoryStream();
                 BinaryWriter writer = new BinaryWriter(output, Encoding.Unicode);
 
                 writer.Write(dispatchId);
                 writer.Write(ProtocolConstants.Success);
 
                 // The response sends the result id, the time the query took, the
                 // total row count, and description of each column in the result.
                 writer.Write(response.ResultId);
                 writer.Write(response.QueryTimeMillis);
                 writer.Write(response.RowCount);
                 int colCount = response.ColumnCount;
                 writer.Write(colCount);
                 for (int i = 0; i < colCount; ++i) {
                     response.GetColumnDescription(i).WriteTo(writer);
                 }
                 writer.Flush();
                 return output.ToArray();
             } catch (DataException e) {
                 //      debug.writeException(e);
                 return Exception(dispatchId, e);
             }
 
         }

byte [] Deveel.Data.Protocol.Processor.ResultSection	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Responds with a part of the result set of a query made via the ProtocolConstants.Query Query.

Parameters

dispatchId	The number we need to respond with.
command

Returns

Definition at line 417 of file Processor.cs.

                                                                      {
             int resultId = ByteBuffer.ReadInt4(command, 8);
             int rowNumber = ByteBuffer.ReadInt4(command, 12);
             int rowCount = ByteBuffer.ReadInt4(command, 16);
 
             try {
                 // Get the result part...
                 ResultPart block = dbInterface.GetResultPart(resultId, rowNumber, rowCount);
 
                 MemoryStream output = new MemoryStream();
                 BinaryWriter writer = new BinaryWriter(output, Encoding.Unicode);
 
                 writer.Write(dispatchId);
                 writer.Write(ProtocolConstants.Success);
 
                 // Send the contents of the result set.
                 // HACK - Work out column count by dividing number of entries in block
                 //   by number of rows.
                 int colCount = block.Count / rowCount;
                 writer.Write(colCount);
                 int bsize = block.Count;
                 for (int index = 0; index < bsize; ++index) {
                     ObjectTransfer.WriteTo(writer, block[index]);
                 }
 
                 writer.Flush();
                 return output.ToArray();
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
         }

abstract void Deveel.Data.Protocol.Processor.SendEvent ( byte[] eventMsg )

protectedpure virtual

Sends an event to the client.

Parameters

eventMsg

This is used to notify the client of trigger events, etc.

Security Issue: This is always invoked by the DatabaseDispatcher. We have to be careful that this method isn't allowed to block. Otherwise the DatabaseDispatcher thread will be out of operation. Unfortunately assuring this may not be possible until non-blocking IO, or we use datagrams for transmission. I know for sure that the TCP implementation is vunrable. If the client doesn't 'read' what we are sending then this'll block when the buffers become full.

static byte [] Deveel.Data.Protocol.Processor.SimpleSuccess ( int dispatchId )

inlinestaticprivate

Creates a response that indicates a simple success of an operation with the given dispatch id.

Parameters

dispatchId

Returns

Definition at line 248 of file Processor.cs.

                                                             {
             byte[] buf = new byte[8];
             ByteBuffer.WriteInteger(dispatchId, buf, 0);
             ByteBuffer.WriteInteger(ProtocolConstants.Success, buf, 4);
             return buf;
         }

static byte [] Deveel.Data.Protocol.Processor.Single ( int val )

inlinestaticprivate

Returns a single 4 byte array with the given int encoded into it.

Parameters

val

Returns

Definition at line 202 of file Processor.cs.

                                               {
             byte[] buf = new byte[4];
             ByteBuffer.WriteInteger(val, buf, 0);
             return buf;
         }

byte [] Deveel.Data.Protocol.Processor.StreamableObjectSection	(	int	dispatchId,
		byte[]	command
	)

inlineprivate

Returns a section of a streamable object.

Parameters

dispatchId	The number we need to respond with.
command

Returns

Definition at line 455 of file Processor.cs.

                                                                                {
             int resultId = ByteBuffer.ReadInt4(command, 8);
             long streamableObjectId = ByteBuffer.ReadInt8(command, 12);
             long offset = ByteBuffer.ReadInt8(command, 20);
             int length = ByteBuffer.ReadInt4(command, 28);
 
             try {
                 byte[] buf = dbInterface.GetStreamableObjectPart(resultId, streamableObjectId, offset, length);
 
                 MemoryStream output = new MemoryStream();
                 BinaryWriter writer = new BinaryWriter(output, Encoding.Unicode);
 
                 writer.Write(dispatchId);
                 writer.Write(ProtocolConstants.Success);
 
                 writer.Write(buf.Length);
                 writer.Write(buf, 0, buf.Length);
                 writer.Flush();
                 return output.ToArray();
             } catch (DataException e) {
                 return Exception(dispatchId, e);
             }
 
         }

Member Data Documentation

int Deveel.Data.Protocol.Processor.authenticationTries

private

Number of authentications tried.

Definition at line 55 of file Processor.cs.

readonly DbController Deveel.Data.Protocol.Processor.controller

private

Definition at line 59 of file Processor.cs.

readonly DatabaseEventCallback Deveel.Data.Protocol.Processor.dbCallback

private

The database call back method that sends database events back to the client.

Definition at line 90 of file Processor.cs.

DatabaseInterface Deveel.Data.Protocol.Processor.dbInterface

private

Definition at line 61 of file Processor.cs.

readonly Dictionary<string, DatabaseInterface> Deveel.Data.Protocol.Processor.dbInterfaces

private

Definition at line 60 of file Processor.cs.

readonly string Deveel.Data.Protocol.Processor.hostString

private

Definition at line 57 of file Processor.cs.

const int Deveel.Data.Protocol.Processor.ServerVersion = 1

private

The version of the server protocol.

Definition at line 45 of file Processor.cs.

ClientConnectionState Deveel.Data.Protocol.Processor.state

private

The current state we are in.

Definition at line 50 of file Processor.cs.

Property Documentation

ClientConnectionState Deveel.Data.Protocol.Processor.ClientState

getprotected

Returns the state of the connection.

0 = not logged in yet. 1 = logged in.

Definition at line 192 of file Processor.cs.

IDatabase Deveel.Data.Protocol.Processor.Database

getprivate set

Definition at line 551 of file Processor.cs.

abstract bool Deveel.Data.Protocol.Processor.IsClosed

get

Returns true if the connection to the client is closed.

Definition at line 549 of file Processor.cs.

The documentation for this class was generated from the following file:

/var/calculate/remote/distfiles/egit-src/deveeldb.git/src/deveeldb/Deveel.Data.Protocol/Processor.cs

Public Member Functions

Protected Member Functions

Properties

Private Member Functions

Static Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation

Property Documentation