Heartbeat + protocol buffer on its way

Header.proto

+package castor;
+
+message HeartBeat {
+ required int64 bytesMoved = 1;
+}

Heartbeat.proto

+package castor;
+
+message HeartBeat {
+ required int64 bytesMoved = 1;
+}

castor/CMakeLists.txt

@@ -25,6 +25,7 @@ cmake_minimum_required (VERSION 2.6)
 if (${COMPILE_SERVER} STREQUAL "1")
   add_subdirectory (db)
   add_subdirectory (gc)
+  add_subdirectory (messages)
   add_subdirectory (rh)
   add_subdirectory (job)
   add_subdirectory (legacymsg)

castor/io/ZMQReactor.hpp

@@ -73,6 +73,8 @@ public:
    * @param timeout Timeout in milliseconds.
    */
   void handleEvents(const int timeout);
+  
+  zmq::context_t& getContext()const {return m_context;}
 private:
     /**
    * Allocates and builds the array of file descriptors to be passed to poll().

castor/messages/CMakeLists.txt

+add_library(castorMessages 
+  Header.pb.cc
+  Heartbeat.pb.cc
+)
+
+target_link_libraries(castorMessages protobuf)
\ No newline at end of file

castor/messages/Constants.hpp

+/******************************************************************************
+ *                      Constants.hpp
+ *
+ * This file is part of the Castor project.
+ * See http://castor.web.cern.ch/castor
+ *
+ * Copyright (C) 2003  CERN
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+ *
+ * 
+ *
+ * @author Castor Dev team, castor-dev@cern.ch
+ *****************************************************************************/
+
+namespace castor {
+  namespace messages{
+    struct protocolType{
+      enum {
+        Tape
+      };
+    };
+    struct reqType{
+      enum {
+        Heartbeat
+      };
+    };
+  }
+}
\ No newline at end of file

castor/messages/Header.pb.h

+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+// source: Header.proto
+
+#ifndef PROTOBUF_Header_2eproto__INCLUDED
+#define PROTOBUF_Header_2eproto__INCLUDED
+
+#include <string>
+
+#include <google/protobuf/stubs/common.h>
+
+#if GOOGLE_PROTOBUF_VERSION < 2003000
+#error This file was generated by a newer version of protoc which is
+#error incompatible with your Protocol Buffer headers.  Please update
+#error your headers.
+#endif
+#if 2003000 < GOOGLE_PROTOBUF_MIN_PROTOC_VERSION
+#error This file was generated by an older version of protoc which is
+#error incompatible with your Protocol Buffer headers.  Please
+#error regenerate this file with a newer version of protoc.
+#endif
+
+#include <google/protobuf/generated_message_util.h>
+#include <google/protobuf/repeated_field.h>
+#include <google/protobuf/extension_set.h>
+#include <google/protobuf/generated_message_reflection.h>
+// @@protoc_insertion_point(includes)
+
+namespace castor {
+namespace messages {
+
+// Internal implementation detail -- do not call these.
+void  protobuf_AddDesc_Header_2eproto();
+void protobuf_AssignDesc_Header_2eproto();
+void protobuf_ShutdownFile_Header_2eproto();
+
+class Header;
+
+// ===================================================================
+
+class Header : public ::google::protobuf::Message {
+ public:
+  Header();
+  virtual ~Header();
+  
+  Header(const Header& from);
+  
+  inline Header& operator=(const Header& from) {
+    CopyFrom(from);
+    return *this;
+  }
+  
+  inline const ::google::protobuf::UnknownFieldSet& unknown_fields() const {
+    return _unknown_fields_;
+  }
+  
+  inline ::google::protobuf::UnknownFieldSet* mutable_unknown_fields() {
+    return &_unknown_fields_;
+  }
+  
+  static const ::google::protobuf::Descriptor* descriptor();
+  static const Header& default_instance();
+  
+  void Swap(Header* other);
+  
+  // implements Message ----------------------------------------------
+  
+  Header* New() const;
+  void CopyFrom(const ::google::protobuf::Message& from);
+  void MergeFrom(const ::google::protobuf::Message& from);
+  void CopyFrom(const Header& from);
+  void MergeFrom(const Header& from);
+  void Clear();
+  bool IsInitialized() const;
+  
+  int ByteSize() const;
+  bool MergePartialFromCodedStream(
+      ::google::protobuf::io::CodedInputStream* input);
+  void SerializeWithCachedSizes(
+      ::google::protobuf::io::CodedOutputStream* output) const;
+  ::google::protobuf::uint8* SerializeWithCachedSizesToArray(::google::protobuf::uint8* output) const;
+  int GetCachedSize() const { return _cached_size_; }
+  private:
+  void SharedCtor();
+  void SharedDtor();
+  void SetCachedSize(int size) const;
+  public:
+  
+  ::google::protobuf::Metadata GetMetadata() const;
+  
+  // nested types ----------------------------------------------------
+  
+  // accessors -------------------------------------------------------
+  
+  // required uint32 magic = 1;
+  inline bool has_magic() const;
+  inline void clear_magic();
+  static const int kMagicFieldNumber = 1;
+  inline ::google::protobuf::uint32 magic() const;
+  inline void set_magic(::google::protobuf::uint32 value);
+  
+  // required uint32 protocolType = 2;
+  inline bool has_protocoltype() const;
+  inline void clear_protocoltype();
+  static const int kProtocolTypeFieldNumber = 2;
+  inline ::google::protobuf::uint32 protocoltype() const;
+  inline void set_protocoltype(::google::protobuf::uint32 value);
+  
+  // required uint32 protocolVersion = 3;
+  inline bool has_protocolversion() const;
+  inline void clear_protocolversion();
+  static const int kProtocolVersionFieldNumber = 3;
+  inline ::google::protobuf::uint32 protocolversion() const;
+  inline void set_protocolversion(::google::protobuf::uint32 value);
+  
+  // required uint32 reqType = 4;
+  inline bool has_reqtype() const;
+  inline void clear_reqtype();
+  static const int kReqTypeFieldNumber = 4;
+  inline ::google::protobuf::uint32 reqtype() const;
+  inline void set_reqtype(::google::protobuf::uint32 value);
+  
+  // required string bodyHashType = 5;
+  inline bool has_bodyhashtype() const;
+  inline void clear_bodyhashtype();
+  static const int kBodyHashTypeFieldNumber = 5;
+  inline const ::std::string& bodyhashtype() const;
+  inline void set_bodyhashtype(const ::std::string& value);
+  inline void set_bodyhashtype(const char* value);
+  inline void set_bodyhashtype(const char* value, size_t size);
+  inline ::std::string* mutable_bodyhashtype();
+  
+  // required string bodyHashValue = 6;
+  inline bool has_bodyhashvalue() const;
+  inline void clear_bodyhashvalue();
+  static const int kBodyHashValueFieldNumber = 6;
+  inline const ::std::string& bodyhashvalue() const;
+  inline void set_bodyhashvalue(const ::std::string& value);
+  inline void set_bodyhashvalue(const char* value);
+  inline void set_bodyhashvalue(const char* value, size_t size);
+  inline ::std::string* mutable_bodyhashvalue();
+  
+  // required string bodySignatureType = 7;
+  inline bool has_bodysignaturetype() const;
+  inline void clear_bodysignaturetype();
+  static const int kBodySignatureTypeFieldNumber = 7;
+  inline const ::std::string& bodysignaturetype() const;
+  inline void set_bodysignaturetype(const ::std::string& value);
+  inline void set_bodysignaturetype(const char* value);
+  inline void set_bodysignaturetype(const char* value, size_t size);
+  inline ::std::string* mutable_bodysignaturetype();
+  
+  // required string bodySignature = 8;
+  inline bool has_bodysignature() const;
+  inline void clear_bodysignature();
+  static const int kBodySignatureFieldNumber = 8;
+  inline const ::std::string& bodysignature() const;
+  inline void set_bodysignature(const ::std::string& value);
+  inline void set_bodysignature(const char* value);
+  inline void set_bodysignature(const char* value, size_t size);
+  inline ::std::string* mutable_bodysignature();
+  
+  // @@protoc_insertion_point(class_scope:castor.messages.Header)
+ private:
+  ::google::protobuf::UnknownFieldSet _unknown_fields_;
+  mutable int _cached_size_;
+  
+  ::google::protobuf::uint32 magic_;
+  ::google::protobuf::uint32 protocoltype_;
+  ::google::protobuf::uint32 protocolversion_;
+  ::google::protobuf::uint32 reqtype_;
+  ::std::string* bodyhashtype_;
+  static const ::std::string _default_bodyhashtype_;
+  ::std::string* bodyhashvalue_;
+  static const ::std::string _default_bodyhashvalue_;
+  ::std::string* bodysignaturetype_;
+  static const ::std::string _default_bodysignaturetype_;
+  ::std::string* bodysignature_;
+  static const ::std::string _default_bodysignature_;
+  friend void  protobuf_AddDesc_Header_2eproto();
+  friend void protobuf_AssignDesc_Header_2eproto();
+  friend void protobuf_ShutdownFile_Header_2eproto();
+  
+  ::google::protobuf::uint32 _has_bits_[(8 + 31) / 32];
+  
+  // WHY DOES & HAVE LOWER PRECEDENCE THAN != !?
+  inline bool _has_bit(int index) const {
+    return (_has_bits_[index / 32] & (1u << (index % 32))) != 0;
+  }
+  inline void _set_bit(int index) {
+    _has_bits_[index / 32] |= (1u << (index % 32));
+  }
+  inline void _clear_bit(int index) {
+    _has_bits_[index / 32] &= ~(1u << (index % 32));
+  }
+  
+  void InitAsDefaultInstance();
+  static Header* default_instance_;
+};
+// ===================================================================
+
+
+// ===================================================================
+
+// Header
+
+// required uint32 magic = 1;
+inline bool Header::has_magic() const {
+  return _has_bit(0);
+}
+inline void Header::clear_magic() {
+  magic_ = 0u;
+  _clear_bit(0);
+}
+inline ::google::protobuf::uint32 Header::magic() const {
+  return magic_;
+}
+inline void Header::set_magic(::google::protobuf::uint32 value) {
+  _set_bit(0);
+  magic_ = value;
+}
+
+// required uint32 protocolType = 2;
+inline bool Header::has_protocoltype() const {
+  return _has_bit(1);
+}
+inline void Header::clear_protocoltype() {
+  protocoltype_ = 0u;
+  _clear_bit(1);
+}
+inline ::google::protobuf::uint32 Header::protocoltype() const {
+  return protocoltype_;
+}
+inline void Header::set_protocoltype(::google::protobuf::uint32 value) {
+  _set_bit(1);
+  protocoltype_ = value;
+}
+
+// required uint32 protocolVersion = 3;
+inline bool Header::has_protocolversion() const {
+  return _has_bit(2);
+}
+inline void Header::clear_protocolversion() {
+  protocolversion_ = 0u;
+  _clear_bit(2);
+}
+inline ::google::protobuf::uint32 Header::protocolversion() const {
+  return protocolversion_;
+}
+inline void Header::set_protocolversion(::google::protobuf::uint32 value) {
+  _set_bit(2);
+  protocolversion_ = value;
+}
+
+// required uint32 reqType = 4;
+inline bool Header::has_reqtype() const {
+  return _has_bit(3);
+}
+inline void Header::clear_reqtype() {
+  reqtype_ = 0u;
+  _clear_bit(3);
+}
+inline ::google::protobuf::uint32 Header::reqtype() const {
+  return reqtype_;
+}
+inline void Header::set_reqtype(::google::protobuf::uint32 value) {
+  _set_bit(3);
+  reqtype_ = value;
+}
+
+// required string bodyHashType = 5;
+inline bool Header::has_bodyhashtype() const {
+  return _has_bit(4);
+}
+inline void Header::clear_bodyhashtype() {
+  if (bodyhashtype_ != &_default_bodyhashtype_) {
+    bodyhashtype_->clear();
+  }
+  _clear_bit(4);
+}
+inline const ::std::string& Header::bodyhashtype() const {
+  return *bodyhashtype_;
+}
+inline void Header::set_bodyhashtype(const ::std::string& value) {
+  _set_bit(4);
+  if (bodyhashtype_ == &_default_bodyhashtype_) {
+    bodyhashtype_ = new ::std::string;
+  }
+  bodyhashtype_->assign(value);
+}
+inline void Header::set_bodyhashtype(const char* value) {
+  _set_bit(4);
+  if (bodyhashtype_ == &_default_bodyhashtype_) {
+    bodyhashtype_ = new ::std::string;
+  }
+  bodyhashtype_->assign(value);
+}
+inline void Header::set_bodyhashtype(const char* value, size_t size) {
+  _set_bit(4);
+  if (bodyhashtype_ == &_default_bodyhashtype_) {
+    bodyhashtype_ = new ::std::string;
+  }
+  bodyhashtype_->assign(reinterpret_cast<const char*>(value), size);
+}
+inline ::std::string* Header::mutable_bodyhashtype() {
+  _set_bit(4);
+  if (bodyhashtype_ == &_default_bodyhashtype_) {
+    bodyhashtype_ = new ::std::string;
+  }
+  return bodyhashtype_;
+}
+
+// required string bodyHashValue = 6;
+inline bool Header::has_bodyhashvalue() const {
+  return _has_bit(5);
+}
+inline void Header::clear_bodyhashvalue() {
+  if (bodyhashvalue_ != &_default_bodyhashvalue_) {
+    bodyhashvalue_->clear();
+  }
+  _clear_bit(5);
+}
+inline const ::std::string& Header::bodyhashvalue() const {
+  return *bodyhashvalue_;
+}
+inline void Header::set_bodyhashvalue(const ::std::string& value) {
+  _set_bit(5);
+  if (bodyhashvalue_ == &_default_bodyhashvalue_) {
+    bodyhashvalue_ = new ::std::string;
+  }
+  bodyhashvalue_->assign(value);
+}
+inline void Header::set_bodyhashvalue(const char* value) {
+  _set_bit(5);
+  if (bodyhashvalue_ == &_default_bodyhashvalue_) {
+    bodyhashvalue_ = new ::std::string;
+  }
+  bodyhashvalue_->assign(value);
+}
+inline void Header::set_bodyhashvalue(const char* value, size_t size) {
+  _set_bit(5);
+  if (bodyhashvalue_ == &_default_bodyhashvalue_) {
+    bodyhashvalue_ = new ::std::string;
+  }
+  bodyhashvalue_->assign(reinterpret_cast<const char*>(value), size);
+}
+inline ::std::string* Header::mutable_bodyhashvalue() {
+  _set_bit(5);
+  if (bodyhashvalue_ == &_default_bodyhashvalue_) {
+    bodyhashvalue_ = new ::std::string;
+  }
+  return bodyhashvalue_;
+}
+
+// required string bodySignatureType = 7;
+inline bool Header::has_bodysignaturetype() const {
+  return _has_bit(6);
+}
+inline void Header::clear_bodysignaturetype() {
+  if (bodysignaturetype_ != &_default_bodysignaturetype_) {
+    bodysignaturetype_->clear();
+  }
+  _clear_bit(6);
+}
+inline const ::std::string& Header::bodysignaturetype() const {
+  return *bodysignaturetype_;
+}
+inline void Header::set_bodysignaturetype(const ::std::string& value) {
+  _set_bit(6);
+  if (bodysignaturetype_ == &_default_bodysignaturetype_) {
+    bodysignaturetype_ = new ::std::string;
+  }
+  bodysignaturetype_->assign(value);
+}
+inline void Header::set_bodysignaturetype(const char* value) {
+  _set_bit(6);
+  if (bodysignaturetype_ == &_default_bodysignaturetype_) {
+    bodysignaturetype_ = new ::std::string;
+  }
+  bodysignaturetype_->assign(value);
+}
+inline void Header::set_bodysignaturetype(const char* value, size_t size) {
+  _set_bit(6);
+  if (bodysignaturetype_ == &_default_bodysignaturetype_) {
+    bodysignaturetype_ = new ::std::string;
+  }
+  bodysignaturetype_->assign(reinterpret_cast<const char*>(value), size);
+}
+inline ::std::string* Header::mutable_bodysignaturetype() {
+  _set_bit(6);
+  if (bodysignaturetype_ == &_default_bodysignaturetype_) {
+    bodysignaturetype_ = new ::std::string;
+  }
+  return bodysignaturetype_;
+}
+
+// required string bodySignature = 8;
+inline bool Header::has_bodysignature() const {
+  return _has_bit(7);
+}
+inline void Header::clear_bodysignature() {
+  if (bodysignature_ != &_default_bodysignature_) {
+    bodysignature_->clear();
+  }
+  _clear_bit(7);
+}
+inline const ::std::string& Header::bodysignature() const {
+  return *bodysignature_;
+}
+inline void Header::set_bodysignature(const ::std::string& value) {
+  _set_bit(7);
+  if (bodysignature_ == &_default_bodysignature_) {
+    bodysignature_ = new ::std::string;
+  }
+  bodysignature_->assign(value);
+}
+inline void Header::set_bodysignature(const char* value) {
+  _set_bit(7);
+  if (bodysignature_ == &_default_bodysignature_) {
+    bodysignature_ = new ::std::string;
+  }
+  bodysignature_->assign(value);
+}
+inline void Header::set_bodysignature(const char* value, size_t size) {
+  _set_bit(7);
+  if (bodysignature_ == &_default_bodysignature_) {
+    bodysignature_ = new ::std::string;
+  }
+  bodysignature_->assign(reinterpret_cast<const char*>(value), size);
+}
+inline ::std::string* Header::mutable_bodysignature() {
+  _set_bit(7);
+  if (bodysignature_ == &_default_bodysignature_) {
+    bodysignature_ = new ::std::string;
+  }
+  return bodysignature_;
+}
+
+
+// @@protoc_insertion_point(namespace_scope)
+
+}  // namespace messages
+}  // namespace castor
+
+#ifndef SWIG
+namespace google {
+namespace protobuf {
+
+
+}  // namespace google
+}  // namespace protobuf
+#endif  // SWIG
+
+// @@protoc_insertion_point(global_scope)
+
+#endif  // PROTOBUF_Header_2eproto__INCLUDED

castor/messages/Header.proto

+package castor.messages;
+
+message Header {
+ required uint32 magic = 1;
+ required uint32 protocolType = 2;
+ required uint32 protocolVersion = 3;
+ required uint32 reqType = 4;
+ required string bodyHashType = 5;		
+ required string bodyHashValue = 6;	
+ required string bodySignatureType= 7;	
+ required string bodySignature = 8;	
+}
\ No newline at end of file

castor/messages/Heartbeat.pb.cc

+// Generated by the protocol buffer compiler.  DO NOT EDIT!
+
+#define INTERNAL_SUPPRESS_PROTOBUF_FIELD_DEPRECATION
+#include "Heartbeat.pb.h"
+#include <google/protobuf/stubs/once.h>
+#include <google/protobuf/io/coded_stream.h>
+#include <google/protobuf/wire_format_lite_inl.h>
+#include <google/protobuf/descriptor.h>
+#include <google/protobuf/reflection_ops.h>
+#include <google/protobuf/wire_format.h>
+// @@protoc_insertion_point(includes)
+
+namespace castor {
+namespace messages {
+
+namespace {
+
+const ::google::protobuf::Descriptor* Heartbeat_descriptor_ = NULL;
+const ::google::protobuf::internal::GeneratedMessageReflection*
+  Heartbeat_reflection_ = NULL;
+
+}  // namespace
+
+
+void protobuf_AssignDesc_Heartbeat_2eproto() {
+  protobuf_AddDesc_Heartbeat_2eproto();
+  const ::google::protobuf::FileDescriptor* file =
+    ::google::protobuf::DescriptorPool::generated_pool()->FindFileByName(
+      "Heartbeat.proto");
+  GOOGLE_CHECK(file != NULL);
+  Heartbeat_descriptor_ = file->message_type(0);
+  static const int Heartbeat_offsets_[1] = {
+    GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(Heartbeat, bytesmoved_),
+  };
+  Heartbeat_reflection_ =
+    new ::google::protobuf::internal::GeneratedMessageReflection(
+      Heartbeat_descriptor_,
+      Heartbeat::default_instance_,
+      Heartbeat_offsets_,
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(Heartbeat, _has_bits_[0]),
+      GOOGLE_PROTOBUF_GENERATED_MESSAGE_FIELD_OFFSET(Heartbeat, _unknown_fields_),
+      -1,
+      ::google::protobuf::DescriptorPool::generated_pool(),
+      ::google::protobuf::MessageFactory::generated_factory(),
+      sizeof(Heartbeat));
+}
+
+namespace {
+
+GOOGLE_PROTOBUF_DECLARE_ONCE(protobuf_AssignDescriptors_once_);
+inline void protobuf_AssignDescriptorsOnce() {
+  ::google::protobuf::GoogleOnceInit(&protobuf_AssignDescriptors_once_,
+                 &protobuf_AssignDesc_Heartbeat_2eproto);
+}
+
+void protobuf_RegisterTypes(const ::std::string&) {
+  protobuf_AssignDescriptorsOnce();
+  ::google::protobuf::MessageFactory::InternalRegisterGeneratedMessage(