aboutsummaryrefslogtreecommitdiffstats
path: root/library/cpp/http/client/fetch/cosocket.h
blob: 8230d36bbe32aa2d31feb9564b74d0142f884567 (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
#pragma once

#include "coctx.h"

#include <library/cpp/coroutine/engine/network.h>
#include <library/cpp/http/fetch_gpl/sockhandler.h>

#include <util/system/error.h>

namespace NHttpFetcher {
    class TCoSocketHandler {
    public:
        TCoSocketHandler() = default;

        ~TCoSocketHandler() {
            Disconnect();
        }

        int Good() const {
            return (Fd != INVALID_SOCKET);
        }

        int Connect(const TAddrList& addrs, TDuration timeout) {
            TCont* cont = CoCtx()->Cont();
            Timeout = timeout;
            for (const auto& item : addrs) {
                try {
                    const sockaddr* sa = item->Addr();
                    TSocketHolder s(NCoro::Socket(sa->sa_family, SOCK_STREAM, 0));
                    if (s.Closed()) {
                        continue;
                    }
                    int err = NCoro::ConnectT(cont, s, sa, item->Len(), Timeout);
                    if (err) {
                        s.Close();
                        errno = err;
                        continue;
                    }
                    SetZeroLinger(s);
                    SetKeepAlive(s, true);
                    Fd.Swap(s);
                    return 0;
                } catch (const TSystemError&) {
                }
            }
            return errno ? errno : EBADF;
        }

        void Disconnect() {
            if (Fd.Closed())
                return;
            try {
                ShutDown(Fd, SHUT_RDWR);
            } catch (const TSystemError&) {
            }
            Fd.Close();
        }

        void shutdown() {
            try {
                ShutDown(Fd, SHUT_WR);
            } catch (TSystemError&) {
            }
        }

        ssize_t send(const void* message, size_t messlen) {
            TCont* cont = CoCtx()->Cont();
            TContIOStatus status = NCoro::WriteT(cont, Fd, message, messlen, Timeout);
            errno = status.Status();
            return status.Status() ? -1 : (ssize_t)status.Processed();
        }

        bool peek() {
            TCont* cont = CoCtx()->Cont();
            if ((errno = NCoro::PollT(cont, Fd, CONT_POLL_READ, Timeout)))
                return false;
            char buf[1];
#ifdef _win32_
            return (1 == ::recv(Fd, buf, 1, MSG_PEEK));
#else
            return (1 == ::recv(Fd, buf, 1, MSG_PEEK | MSG_DONTWAIT));
#endif
        }

        ssize_t read(void* message, size_t messlen) {
            TCont* cont = CoCtx()->Cont();
            TContIOStatus status = NCoro::ReadT(cont, Fd, message, messlen, Timeout);
            errno = status.Status();
            return status.Status() ? -1 : (ssize_t)status.Processed();
        }

    protected:
        TSocketHolder Fd;
        TDuration Timeout;
        static THolder<TIpAddress> AddrToBind;
    };
}