Begin stratum implementation

src/server/poolserver/CMakeLists.txt

@@ -2,23 +2,26 @@
 # Add sources
 file(GLOB_RECURSE sources_Server Server/*.cpp Server/*.h)
 file(GLOB_RECURSE sources_Database Database/*.cpp Database/*.h)
+file(GLOB_RECURSE sources_Stratum Stratum/*.cpp Stratum/*.h)
 file(GLOB sources_localdir *.cpp *.h)
 
 set(sources_Poolserver
   ${sources_Server}
   ${sources_Database}
+  ${sources_Stratum}
   ${sources_localdir}
 )
 
 include_directories(
+  ${CMAKE_CURRENT_SOURCE_DIR}
+  ${CMAKE_CURRENT_SOURCE_DIR}/Server
+  ${CMAKE_CURRENT_SOURCE_DIR}/Database
   ${CMAKE_SOURCE_DIR}/src/server/shared
   ${CMAKE_SOURCE_DIR}/src/server/shared/Configuration
   ${CMAKE_SOURCE_DIR}/src/server/shared/MySQL
   ${CMAKE_SOURCE_DIR}/src/server/shared/Logging
   ${CMAKE_SOURCE_DIR}/src/server/shared/JSON
   ${CMAKE_SOURCE_DIR}/src/server/shared/JSONRPC
-  ${CMAKE_SOURCE_DIR}/src/server/poolserver/Server
-  ${CMAKE_SOURCE_DIR}/src/server/poolserver/Database
   ${Boost_INCLUDE_DIR}
   ${MYSQL_INCLUDE_DIR}
 )

src/server/poolserver/Main.cpp

@@ -40,6 +40,7 @@ bool InitConfig(int argc, char *argv[])
     descStratum.add_options()
         ("StratumHost,sh", boost::program_options::value<std::string>()->default_value("0.0.0.0"), "Stratum server host")
         ("StratumPort,sp", boost::program_options::value<uint16_t>()->default_value(3333), "Stratum server port")
+        ("StratumBlockCheckTime", boost::program_options::value<uint32>()->default_value(2000), "Time between block checks in ms")
     ;
     
     // Logging

src/server/poolserver/Server/Server.cpp

@@ -1,5 +1,6 @@
 #include "Server.h"
 #include "Config.h"
+#include "JSONRPC.h"
 #include "Log.h"
 #include "Stratum/Server.h"
 #include "ServerDatabaseEnv.h"
@@ -29,7 +30,30 @@ int Server::Run()
 {
     sLog.Info(LOG_SERVER, "Server is starting...");
     
-    InitDatabase();
+    //InitDatabase();
+    
+    // Main io service
+    asio::io_service io_service;
+    
+    Stratum::Server srv(io_service);
+    
+    // Init Bitcoin RPC
+    JSONRPCConnectionInfo coninfo;
+    coninfo.Host = "84.240.15.208";
+    coninfo.Port = "9902";
+    coninfo.User = "ppcuser";
+    coninfo.Pass = "DYAL6bC4RUHksL6ikdx7";
+    
+    JSONRPC* bitcoinrpc = new JSONRPC();
+    bitcoinrpc->Connect(coninfo);
+    srv.SetBitcoinRPC(bitcoinrpc);
+    
+    // Start stratum server
+    tcp::endpoint endpoint(tcp::v4(), sConfig.Get<uint16>("StratumPort"));
+    srv.Start(endpoint);
+    
+    io_service.run();
+    
     
     //sDatabase.Execute("INSERT INTO `test_table` VALUES ('999', 'sync', '1.1')");
     //sDatabase.ExecuteAsync("INSERT INTO `test_table` VALUES ('999', 'sync', '1.1')");
@@ -40,19 +64,19 @@ int Server::Run()
     stmt->SetFloat(2, 5.987);
     sDatabase.ExecuteAsync(stmt);*/
     
-    MySQL::PreparedStatement* stmt = sDatabase.GetPreparedStatement(STMT_QUERY_TEST_TABLE);
+    //MySQL::PreparedStatement* stmt = sDatabase.GetPreparedStatement(STMT_QUERY_TEST_TABLE);
-    MySQL::QueryResult result = sDatabase.Query(stmt);
+    //MySQL::QueryResult result = sDatabase.Query(stmt);
     
     
     //sDatabase.QueryAsync("SELECT * FROM `test_table`", &AsyncQueryCallback);
     //MySQL::QueryResult result = sDatabase.Query("SELECT * FROM `test_table`");
-    if (result) {
+    /*if (result) {
         sLog.Info(LOG_SERVER, "Metadata: F: %u R: %u", result->GetFieldCount(), result->GetRowCount());
         while (MySQL::Field* fields = result->FetchRow()) {
             sLog.Info(LOG_SERVER, "Row: %i %s", fields[0].GetUInt32(), fields[1].GetString().c_str());
         }
     } else
-        sLog.Info(LOG_SERVER, "Empty result");
+        sLog.Info(LOG_SERVER, "Empty result");*/
     
     // Start stratum server
     sLog.Info(LOG_SERVER, "Starting stratum");

src/server/poolserver/Stratum/Client.h

@@ -0,0 +1,65 @@
+#ifndef STRATUM_CLIENT_H_
+#define STRATUM_CLIENT_H_
+
+#include "Common.h"
+#include "Log.h"
+#include "JSON.h"
+
+#include <boost/asio.hpp>
+#include <boost/bind.hpp>
+
+#define MAX_PACKET 2048
+
+using namespace boost;
+using namespace boost::asio::ip;
+
+namespace Stratum
+{
+    class Client
+    {
+    public:
+        Client(asio::io_service& io_service) : _socket(io_service)
+        {
+        }
+        
+        tcp::socket& GetSocket()
+        {
+            return _socket;
+        }
+        
+        void Start()
+        {
+            boost::asio::async_read_until(
+                _socket,
+                _recvBuffer,
+                '\n',
+                boost::bind(&Client::_OnReceive, this, asio::placeholders::error, asio::placeholders::bytes_transferred));
+        }
+        
+        void SendMessage(JSON msg)
+        {
+            std::string data = msg.ToString();
+            _socket.send(boost::asio::buffer(data.c_str(), data.length()));
+        }
+        
+        void OnMessage(JSON msg)
+        {
+            std::string method = msg.Get<std::string>("method");
+            sLog.Debug(LOG_SERVER, "Method: %s", method.c_str());
+        }
+    public:
+        void _OnReceive(const boost::system::error_code& error, size_t bytes_transferred)
+        {
+            std::istream is(&_recvBuffer);
+            std::stringstream iss;
+            iss << is.rdbuf();
+            sLog.Debug(LOG_SERVER, "Received: %s", iss.str().c_str());
+            OnMessage(JSON::FromString(iss.str()));
+        }
+    private:
+        asio::streambuf _recvBuffer;
+        tcp::socket _socket;
+    };
+}
+
+#endif

src/server/poolserver/Stratum/Server.h

@@ -0,0 +1,122 @@
+#ifndef STRATUM_SERVER_H_
+#define STRATUM_SERVER_H_
+
+#include "Common.h"
+#include "Client.h"
+#include "Config.h"
+#include "Log.h"
+#include "JSON.h"
+#include "JSONRPC.h"
+
+#include <boost/asio.hpp>
+#include <boost/bind.hpp>
+#include <list>
+
+using namespace boost;
+using namespace boost::asio::ip;
+
+namespace Stratum
+{
+    struct ListenInfo
+    {
+        std::string Host;
+        uint16 Port;
+    };
+    
+    class Server
+    {
+    public:
+        Server(asio::io_service& io_service) : _io_service(io_service), _acceptor(io_service), _blockCheckTimer(io_service), _blockHeight(0)
+        {
+        }
+        
+        ~Server()
+        {
+            std::list<Client*>::iterator it;
+            for (it = _clients.begin(); it != _clients.end(); ++it)
+                delete *it;
+        }
+        
+        void Start(tcp::endpoint endpoint)
+        {
+            _CheckBlocks();
+            
+            _acceptor.open(endpoint.protocol());
+            _acceptor.set_option(tcp::acceptor::reuse_address(true));
+            _acceptor.bind(endpoint);
+            _acceptor.listen();
+            
+            _StartAccept();
+            
+            sLog.Debug(LOG_STRATUM, "Stratum server started");
+        }
+        
+        void SetBitcoinRPC(JSONRPC* bitcoinrpc)
+        {
+            _bitcoinrpc = bitcoinrpc;
+        }
+        
+        void SendToAll(JSON msg)
+        {
+            std::list<Client*>::iterator it;
+            for (it = _clients.begin(); it != _clients.end(); ++it)
+                (*it)->SendMessage(msg);
+        }
+        
+    private:
+        void _StartAccept()
+        {
+            Client* client = new Client(_io_service);
+            
+            _acceptor.async_accept(client->GetSocket(), boost::bind(&Server::_OnAccept, this, client, asio::placeholders::error));
+        }
+        
+        void _OnAccept(Client* client, const boost::system::error_code& error)
+        {
+            if (!error) {
+                sLog.Debug(LOG_STRATUM, "New stratum client accepted");
+                client->Start();
+                _clients.push_back(client);
+            } else {
+                sLog.Debug(LOG_STRATUM, "Failed to accept stratum client");
+                delete client;
+            }
+            
+            _StartAccept();
+        }
+        
+        void _CheckBlocks()
+        {
+            sLog.Debug(LOG_STRATUM, "Checking for new blocks...");
+            
+            JSON response = _bitcoinrpc->Query("getinfo");
+            uint32 curBlock = response.Get<uint32>("blocks");
+            
+            // Initializing server
+            if (_blockHeight == 0) {
+                _blockHeight = curBlock;
+            } else if (curBlock > _blockHeight) {
+                sLog.Debug(LOG_STRATUM, "New block on network! Height: %u", curBlock);
+                _blockHeight = curBlock;
+                // do some crazy stuff
+            }
+            
+            _blockCheckTimer.expires_from_now(boost::posix_time::milliseconds(sConfig.Get<uint32>("StratumBlockCheckTime")));
+            _blockCheckTimer.async_wait(boost::bind(&Server::_CheckBlocks, this));
+        }
+    private:
+        // Network
+        std::list<Client*> _clients;
+        asio::io_service& _io_service;
+        tcp::acceptor _acceptor;
+        
+        // RPC
+        JSONRPC* _bitcoinrpc;
+        
+        // Bitcoin info
+        asio::deadline_timer _blockCheckTimer;
+        uint32 _blockHeight;
+    };
+}
+
+#endif

src/server/shared/Bitcoin/BitCoinRPC.h

@@ -1,11 +0,0 @@
-#ifndef BITCOINRPC_H_
-#define BITCOINRPC_H_
-
-#include "JSONRPC.h"
-
-class BitCoinRPC: public JSONRPC
-{
-    
-};
-
-#endif

src/server/shared/Bitcoin/Block.h

@@ -0,0 +1,40 @@
+#ifndef BITCOIN_H_
+#define BITCOIN_H_
+
+namespace Bitcoin
+{
+    class TxIn
+    {
+    };
+    
+    class TxOut
+    {
+    };
+    
+    class Transaction
+    {
+        uint32 Version;
+        std::vector<TxIn> in;
+        std::vector<TxOut> out;
+        uint32 LockTime;
+    };
+    
+    class BlockHeader
+    {
+        uint32 Version;
+        std::array<char, 32> PrevBlockHash;
+        std::array<char, 32> MerkleRootHash;
+        uint32 Time;
+        uint32 Bits;
+        uint32 Nonce;
+    };
+    
+    class Block : public BlockHeader
+    {
+        std::vector<Transaction> tx;
+    };
+    
+    class BlockTemplate
+}
+
+#endif

src/server/shared/CMakeLists.txt

@@ -2,7 +2,6 @@
 file(GLOB_RECURSE sources_Configuration Configuration/*.cpp Configuration/*.h)
 file(GLOB_RECURSE sources_MySQL MySQL/*.cpp MySQL/*.h)
 file(GLOB_RECURSE sources_Logging Logging/*.cpp Logging/*.h)
-file(GLOB_RECURSE sources_Stratum Stratum/*.cpp Stratum/*.h)
 file(GLOB_RECURSE sources_JSON JSON/*.cpp JSON/*.h)
 file(GLOB_RECURSE sources_JSONRPC JSONRPC/*.cpp JSONRPC/*.h)
 
@@ -12,7 +11,6 @@ set(sources_Shared
   ${sources_Configuration}
   ${sources_MySQL}
   ${sources_Logging}
-  ${sources_Stratum}
   ${sources_JSON}
   ${sources_JSONRPC}
   ${sources_localdir}

src/server/shared/JSONRPC/JSONRPC.cpp

@@ -5,6 +5,8 @@
 #include "Util.h"
 #include "Log.h"
 
+using namespace boost::asio::ip;
+
 bool JSONRPC::Connect(JSONRPCConnectionInfo connInfo)
 {
     _connInfo = connInfo;
@@ -14,17 +16,18 @@ bool JSONRPC::Connect(JSONRPCConnectionInfo connInfo)
     
     sLog.Debug(LOG_JSONRPC, "JSONRPC::Connect(): B64Auth: %s", _connInfo.B64Auth.c_str());
 
-    boost::asio::ip::tcp::resolver resolver(_ios);
+    tcp::resolver resolver(_ios);
-    boost::asio::ip::tcp::resolver::query q(_connInfo.Host, _connInfo.Port);
+    tcp::resolver::query q(tcp::v4(), connInfo.Host, connInfo.Port);
-    boost::asio::ip::tcp::resolver::iterator epi = resolver.resolve(q);
+    tcp::resolver::iterator epi = resolver.resolve(q);
-    boost::asio::ip::tcp::resolver::iterator end;
+    tcp::resolver::iterator end;
     
     boost::system::error_code error = boost::asio::error::host_not_found;
     
     while (error && epi != end)
     {
+        _ep = *epi++;
         _sock.close();
-        _sock.connect(*epi, error);
+        _sock.connect(_ep, error);
     }
     
     if (error)
@@ -33,8 +36,6 @@ bool JSONRPC::Connect(JSONRPCConnectionInfo connInfo)
         return false;
     }
     
-    _ep = *epi;
-    
     _sock.close();
 
     return true;
@@ -123,5 +124,7 @@ JSON JSONRPC::Query(std::string method, JSON params)
     
     sLog.Debug(LOG_JSONRPC, "JSONRPC::Query(): JSON Response: %s", jsonresponse.c_str());
     
-    return JSON::FromString(jsonresponse);
+    JSON data = JSON::FromString(jsonresponse);
+    
+    return data["result"];
 }

src/server/shared/Logging/Log.h

@@ -18,7 +18,8 @@ enum LogType
     LOG_SERVER                          = 1,
     LOG_DATABASE                        = 2,
     LOG_JSON                            = 3,
-    LOG_JSONRPC                         = 4
+    LOG_JSONRPC                         = 4,
+    LOG_STRATUM                         = 5,
 };
 
 enum LogLevel

src/server/shared/MySQL/DatabaseConnection.cpp

@@ -281,8 +281,8 @@ namespace MySQL
         if (index >= _stmts.size())
             _stmts.resize(index+1);
         
-        ConnectionPreparedStatement* mStmt = new ConnectionPreparedStatement(stmt);
+        ConnectionPreparedStatement* cstmt = new ConnectionPreparedStatement(stmt);
-        _stmts[index] = mStmt;
+        _stmts[index] = cstmt;
         
         sLog.Debug(LOG_DATABASE, "Prepared STMT id: %u, sql: \"%s\"", index, sql);
         

src/server/shared/Stratum/Server.cpp

@@ -1,6 +0,0 @@
-#include "Server.h"
-
-Stratum::Server::Server(std::string ip, uint32_t port)
-{
-
-}

src/server/shared/Stratum/Server.h

@@ -1,20 +0,0 @@
-#ifndef STRATUM_SERVER_H_
-#define STRATUM_SERVER_H_
-
-#include <boost/cstdint.hpp>
-#include <string>
-
-namespace Stratum
-{
-    class Server
-    {
-    public:
-        Server(std::string ip, uint32_t port);
-        ~Server();
-
-    private:
-
-    };
-}
-
-#endif