sozu 0.14.2

sozu, a fast, reliable, hot reconfigurable HTTP reverse proxy
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

use std::{collections::VecDeque, fmt, os::unix::io::AsRawFd};

use futures::SinkExt;
use libc::pid_t;
use nix::{sys::signal::kill, unistd::Pid};

use sozu_command_lib::{
    channel::Channel,
    command::{RunState, WorkerInfo},
    config::Config,
    proxy::{ProxyRequest, ProxyRequestOrder, ProxyResponse},
    scm_socket::ScmSocket,
};

pub struct Worker {
    pub id: u32,
    /// for the worker to receive requests and respond to the main process
    pub worker_channel: Option<Channel<ProxyRequest, ProxyResponse>>,
    /// file descriptor of the command channel
    pub worker_channel_fd: i32,
    pub pid: pid_t,
    pub run_state: RunState,
    pub queue: VecDeque<ProxyRequest>,
    /// Used to send and receive listeners (socket addresses and file descriptors)
    pub scm_socket: ScmSocket,
    pub sender: Option<futures::channel::mpsc::Sender<ProxyRequest>>,
}

impl Worker {
    pub fn new(
        id: u32,
        pid: pid_t,
        command_channel: Channel<ProxyRequest, ProxyResponse>,
        scm_socket: ScmSocket,
        _: &Config,
    ) -> Worker {
        Worker {
            id,
            worker_channel_fd: command_channel.sock.as_raw_fd(),
            worker_channel: Some(command_channel),
            sender: None,
            pid,
            run_state: RunState::Running,
            queue: VecDeque::new(),
            scm_socket,
        }
    }

    /// send proxy request to the worker, via the mpsc sender
    pub async fn send(&mut self, request_id: String, data: ProxyRequestOrder) {
        if let Some(worker_tx) = self.sender.as_mut() {
            if let Err(e) = worker_tx
                .send(ProxyRequest {
                    id: request_id.clone(),
                    order: data,
                })
                .await
            {
                error!(
                    "error sending message {} to worker {:?}: {:?}",
                    request_id, self.id, e
                );
            }
        }
    }

    pub fn the_pid_is_alive(&self) -> bool {
        // send a kill -0 to check on the pid, if it's dead it should be an error
        kill(Pid::from_raw(self.pid), None).is_ok()
    }

    pub fn info(&self) -> WorkerInfo {
        WorkerInfo {
            id: self.id,
            pid: self.pid,
            run_state: self.run_state,
        }
    }

    /*
    pub fn push_message(&mut self, message: ProxyRequest) {
      self.queue.push_back(message);
      self.channel.interest.insert(Ready::writable());
    }

    pub fn can_handle_events(&self) -> bool {
      self.channel.readiness().is_readable() || (!self.queue.is_empty() && self.channel.readiness().is_writable())
    }*/
}

impl fmt::Debug for Worker {
    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
        write!(
            f,
            "Worker {{ id: {}, run_state: {:?} }}",
            self.id, self.run_state
        )
    }
}