aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/messagebus/message.h
blob: 005ca10c652919e2ab4077c07d03e52f8f23dbcf (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
#pragma once

#include "base.h"
#include "local_flags.h"
#include "message_status.h"
#include "netaddr.h"
#include "socket_addr.h"

#include <util/generic/array_ref.h>
#include <util/generic/noncopyable.h>
#include <util/generic/ptr.h>
#include <util/generic/string.h>
#include <util/system/defaults.h>
#include <util/system/type_name.h>
#include <util/system/yassert.h>

#include <optional>
#include <typeinfo>

namespace NBus {
    ///////////////////////////////////////////////////////////////////
    /// \brief Structure to preserve identity from message to reply
    struct TBusIdentity : TNonCopyable {
        friend class TBusMessage;
        friend class NPrivate::TRemoteServerSession;
        friend struct NPrivate::TClientRequestImpl;
        friend class TOnMessageContext;

        // TODO: make private
        TBusKey MessageId;

    private:
        ui32 Size;
        TIntrusivePtr<NPrivate::TRemoteServerConnection> Connection;
        ui16 Flags;
        ui32 LocalFlags;
        TInstant RecvTime;

#ifndef NDEBUG
        std::optional<TString> MessageType;
#endif

    private:
        // TODO: drop
        TNetAddr GetNetAddr() const;

    public:
        void Pack(char* dest);
        void Unpack(const char* src);

        bool IsInWork() const {
            return LocalFlags & NPrivate::MESSAGE_IN_WORK;
        }

        // for internal use only
        void BeginWork() {
            SetInWork(true);
        }

        // for internal use only
        void EndWork() {
            SetInWork(false);
        }

        TBusIdentity();
        ~TBusIdentity();

        void Swap(TBusIdentity& that) {
            DoSwap(MessageId, that.MessageId);
            DoSwap(Size, that.Size);
            DoSwap(Connection, that.Connection);
            DoSwap(Flags, that.Flags);
            DoSwap(LocalFlags, that.LocalFlags);
            DoSwap(RecvTime, that.RecvTime);
#ifndef NDEBUG
            DoSwap(MessageType, that.MessageType);
#endif
        }

        TString ToString() const;

    private:
        void SetInWork(bool inWork) {
            if (LocalFlags == 0 && inWork) {
                LocalFlags = NPrivate::MESSAGE_IN_WORK;
            } else if (LocalFlags == NPrivate::MESSAGE_IN_WORK && !inWork) {
                LocalFlags = 0;
            } else {
                Y_FAIL("impossible combination of flag and parameter: %s %d",
                       inWork ? "true" : "false", unsigned(LocalFlags));
            }
        }

        void SetMessageType(const std::type_info& messageTypeInfo) {
#ifndef NDEBUG
            Y_VERIFY(!MessageType, "state check");
            MessageType = TypeName(messageTypeInfo);
#else
            Y_UNUSED(messageTypeInfo);
#endif
        }
    };

    static const size_t BUS_IDENTITY_PACKED_SIZE = sizeof(TBusIdentity);

    ///////////////////////////////////////////////////////////////
    /// \brief Message flags in TBusHeader.Flags
    enum EMessageFlags {
        MESSAGE_COMPRESS_INTERNAL = 0x8000, ///< message is compressed
        MESSAGE_COMPRESS_RESPONSE = 0x4000, ///< message prefers compressed response
        MESSAGE_VERSION_INTERNAL = 0x00F0,  ///< these bits are used as version
    };

//////////////////////////////////////////////////////////
/// \brief Message header present in all message send and received

/// This header is send into the wire.
/// \todo fix for low/high end, 32/64bit some day
#pragma pack(1)
    struct TBusHeader {
        friend class TBusMessage;

        TBusKey Id = 0;           ///< unique message ID
        ui32 Size = 0;            ///< total size of the message
        TBusInstant SendTime = 0; ///< time the message was sent
        ui16 FlagsInternal = 0;   ///< TRACE is one of the flags
        ui16 Type = 0;            ///< to be used by TBusProtocol

        int GetVersionInternal() {
            return (FlagsInternal & MESSAGE_VERSION_INTERNAL) >> 4;
        }
        void SetVersionInternal(unsigned ver = YBUS_VERSION) {
            FlagsInternal |= (ver << 4);
        }

    public:
        TBusHeader() {
        }
        TBusHeader(TArrayRef<const char> data) {
            ReadHeader(data);
        }

    private:
        /// function for serialization/deserialization of the header
        /// returns number of bytes written/read
        int ReadHeader(TArrayRef<const char> data);

        void GenerateId();
    };
#pragma pack()

#define TBUSMAX_MESSAGE 26 * 1024 * 1024 + sizeof(NBus::TBusHeader) ///< is't it enough?
#define TBUSMIN_MESSAGE sizeof(NBus::TBusHeader)                    ///< can't be less then header

    inline bool IsVersionNegotiation(const NBus::TBusHeader& header) {
        return header.Id == 0 && header.Size == sizeof(TBusHeader);
    }

    //////////////////////////////////////////////////////////
    /// \brief Base class for all messages passed in the system

    enum ECreateUninitialized {
        MESSAGE_CREATE_UNINITIALIZED,
    };

    class TBusMessage
       : protected  TBusHeader,
          public TRefCounted<TBusMessage, TAtomicCounter, TDelete>,
          private TNonCopyable {
        friend class TLocalSession;
        friend struct ::NBus::NPrivate::TBusSessionImpl;
        friend class ::NBus::NPrivate::TRemoteServerSession;
        friend class ::NBus::NPrivate::TRemoteClientSession;
        friend class ::NBus::NPrivate::TRemoteConnection;
        friend class ::NBus::NPrivate::TRemoteClientConnection;
        friend class ::NBus::NPrivate::TRemoteServerConnection;
        friend struct ::NBus::NPrivate::TBusMessagePtrAndHeader;

    private:
        ui32 LocalFlags;

        /// connection identity for reply set by PushMessage()
        NPrivate::TBusSocketAddr ReplyTo;
        // server-side response only, hack
        ui32 RequestSize;

        TInstant RecvTime;

    public:
        /// constructor to create messages on sending end
        TBusMessage(ui16 type, int approxsize = sizeof(TBusHeader));

        /// constructor with serialzed data to examine the header
        TBusMessage(ECreateUninitialized);

        // slow, for diagnostics only
        virtual TString Describe() const;

        // must be called if this message object needs to be reused
        void Reset();

        void CheckClean() const;

        void SetCompressed(bool);
        void SetCompressedResponse(bool);

    private:
        bool IsCompressed() const {
            return FlagsInternal & MESSAGE_COMPRESS_INTERNAL;
        }
        bool IsCompressedResponse() const {
            return FlagsInternal & MESSAGE_COMPRESS_RESPONSE;
        }

    public:
        /// can have private data to destroy
        virtual ~TBusMessage();

        /// returns header of the message
        TBusHeader* GetHeader() {
            return this;
        }
        const TBusHeader* GetHeader() const {
            return this;
        }

        /// helper to return type for protocol object to unpack object
        static ui16 GetType(TArrayRef<const char> data) {
            return TBusHeader(data).Type;
        }

        /// returns payload data
        static TArrayRef<const char> GetPayload(TArrayRef<const char> data) {
            return data.Slice(sizeof(TBusHeader));
        }

    private:
        void DoReset();

        /// serialize message identity to be used to construct reply message
        void GetIdentity(TBusIdentity& ident) const;

        /// set message identity from serialized form
        void SetIdentity(const TBusIdentity& ident);

    public:
        TNetAddr GetReplyTo() const {
            return ReplyTo.ToNetAddr();
        }

        /// store of application specific data, never serialized into wire
        void* Data;
    };

    class TBusMessageAutoPtr: public TAutoPtr<TBusMessage> {
    public:
        TBusMessageAutoPtr() {
        }

        TBusMessageAutoPtr(TBusMessage* message)
            : TAutoPtr<TBusMessage>(message)
        {
        }

        template <typename T1>
        TBusMessageAutoPtr(const TAutoPtr<T1>& that)
            : TAutoPtr<TBusMessage>(that.Release())
        {
        }
    };

}